Getting Started with the Claude Agent SDK

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

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"https://www.kdnuggets.com/"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())

/trend "AI chip startups"