Getting started with the Claude Agent SDK

irm  | iex

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

pip install claude-agent-sdk

import Asyncio import time import OS NOS ENTIVETE from DateTime TetTeme from PathLib import method ( Claudeasuent Helper , ClaudesdkClient , ClaudesSsuent , TextMessage , Tool   Quick Scan of Multi source (Auto-Satas Markdown) / Scan    Short scan of one page / Help / Logout / Your logout / Quit ""[90m"
RESET = "33[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 "def _esuwa_safe (method: PAST) without writing If directory / permission fails back to ~ / trendsmith / reports. "" "" fb.write_text(text, encoding = "UTF-8") Return FB def save

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 nn" + 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())

/trend "AI chip startups"