Claude Agent SDK入门指南

irm https://claude.ai/install.ps1 | iex

C:\Users\\.local\bin

claude --version

npm i -g @anthropic-ai/claude-code

pip install claude-agent-sdk

import asyncio import os import re import sys import time from datetime import datetime from pathlib import Path from claude_agent_sdk import ( AssistantMessage, ClaudeAgentOptions, ClaudeSDKClient, ResultMessage, TextBlock, ToolResultBlock, ToolUseBlock, ) HELP = """Commands: /trend  Quick multi-source scan (auto-saves markdown) /scan  Short one-page scan /help /exit Help / Quit """ MODEL = os.getenv("CLAUDE_MODEL", "sonnet") # e.g. "sonnet-4.5" GRAY = "\033[90m" RESET = "\033[0m"

SYS = """You are TrendSmith, a fast, concise trend researcher. - Finish quickly (~20 s). - For /trend: ≤1 WebSearch + ≤2 WebFetch from distinct domains. - For /scan: ≤1 WebFetch only. Return for /trend: TL;DR (1 line) 3-5 Signals (short bullets) Key Players, Risks, 30/90-day Watchlist Sources (markdown: **Title** -- URL) Return for /scan: 5 bullets + TL;DR + Sources. After finishing /trend, the client will auto-save your full brief. """ BASE = Path(__file__).parent REPORTS = BASE / "reports"

def _ts(): return datetime.now().strftime("%Y%m%d_%H%M") def _sanitize(s: str): return re.sub(r"[^\w\-.]+", "_", s).strip("_") or "untitled" def _ensure_dir(p: Path): try: p.mkdir(parents=True, exist_ok=True) except Exception: pass def _safe_write(path: Path, text: str) -> Path: """Write text to path; if directory/permission fails, fall back to ~/TrendSmith/reports.""" try: _ensure_dir(path.parent) path.write_text(text, encoding="utf-8") return path except Exception: home_reports = Path.home() / "TrendSmith" / "reports" _ensure_dir(home_reports) fb = home_reports / path.name fb.write_text(text, encoding="utf-8") return fb def save_report(topic: str, text: str) -> Path: filename = f"{_sanitize(topic)}_{_ts()}.md" target = REPORTS / filename return _safe_write(target, text.strip() + "\n")

class State: def __init__(self): self.brief = "" self.model_raw = None self.usage = {} self.cost = None self.last_cmd = None self.last_topic = None self.tools = {} self.t0 = 0.0 self.t1 = 0.0 def reset(self): self.brief = "" self.model_raw = None self.usage = {} self.cost = None self.tools = {} self.t0 = time.perf_counter() self.t1 = 0.0 def friendly_model(name: str | None) -> str: if not name: return MODEL n = (name or "").lower() if "sonnet-4-5" in n or "sonnet_4_5" in n: return "Claude 4.5 Sonnet" if "sonnet" in n: return "Claude Sonnet" if "haiku" in n: return "Claude Haiku" if "opus" in n: return "Claude Opus" return name or "Unknown"

def usage_footer(st: State, opts_model: str): st.t1 = st.t1 or time.perf_counter() dur = st.t1 - st.t0 usage = st.usage or {} it = usage.get("input_tokens") ot = usage.get("output_tokens") total = usage.get("total_tokens") if total is None and (it is not None or ot is not None): total = (it or 0) + (ot or 0) tools_used = ", ".join(f"{k}×{v}" for k, v in st.tools.items()) or "--" model_label = friendly_model(st.model_raw or opts_model) box = [ "┌─ Run Summary ─────────────────────────────────────────────", f"│ Model: {model_label}", f"│ Tokens: {total if total is not None else '?'}" + (f" (in={it if it is not None else '?'} | out={ot if ot is not None else '?}')" if (it is not None or ot is not None) else ""), f"│ Tools: {tools_used}", f"│ Duration: {dur:.1f}s", "└───────────────────────────────────────────────────────────", ] print(GRAY + "\n".join(box) + RESET, file=sys.stderr)

async def main(): """Setup → REPL → parse → query/stream → auto-save → summary.""" st = State() _ensure_dir(REPORTS) opts = ClaudeAgentOptions( model=MODEL, system_prompt=SYS, allowed_tools=["WebFetch", "WebSearch"], ) print("📈 TrendSmith \n\n" + HELP) async with ClaudeSDKClient(options=opts) as client: while True: # Read input try: user = input("\nYou: ").strip() except (EOFError, KeyboardInterrupt): print("\nBye!") break if not user: continue low = user.lower() # Basic commands if low in {"/exit", "exit", "quit"}: print("Bye!") break if low in {"/help", "help"}: print(HELP) continue # Parse into a prompt if low.startswith("/trend "): topic = user.split(" ", 1)[1].strip().strip('"') if not topic: print('e.g. /trend "AI chip startups"') continue st.last_cmd, st.last_topic = "trend", topic prompt = f"Run a fast trend scan for '{topic}' following the output spec." elif low.startswith("/scan "): q = user.split(" ", 1)[1].strip() if not q: print('e.g. /scan "AI hardware news"') continue st.last_cmd, st.last_topic = "scan", q prompt = f"Quick scan for '{q}' in under 10s (≤1 WebFetch). Return 5 bullets + TL;DR + sources." else: st.last_cmd, st.last_topic = "free", None prompt = user # Execute request and stream results st.reset() print(f"{GRAY}▶ Working...{RESET}") try: await client.query(prompt) except Exception as e: print(f"{GRAY}❌ Query error: {e}{RESET}") continue try: async for m in client.receive_response(): if isinstance(m, AssistantMessage): st.model_raw = st.model_raw or m.model for b in m.content: if isinstance(b, TextBlock): st.brief += b.text or "" print(b.text or "", end="") elif isinstance(b, ToolUseBlock): name = b.name or "Tool" st.tools[name] = st.tools.get(name, 0) + 1 print(f"{GRAY}\n🛠 Tool: {name}{RESET}") elif isinstance(b, ToolResultBlock): pass # quiet tool payloads elif isinstance(m, ResultMessage): st.usage = m.usage or {} st.cost = m.total_cost_usd except Exception as e: print(f"{GRAY}\n⚠ Stream error: {e}{RESET}") # Auto-save trend briefs and show the summary if st.last_cmd == "trend" and st.brief.strip(): try: saved_path = save_report(st.last_topic or "trend", st.brief) print(f"\n{GRAY}✅ Auto-saved → {saved_path}{RESET}") except Exception as e: print(f"{GRAY}⚠ Save error: {e}{RESET}") st.t1 = time.perf_counter() usage_footer(st, opts.model) if __name__ == "__main__": asyncio.run(main())

python .\trend_smith.py

/trend "AI chip startups"

/scan Code CLI tools