Getting Began with the Claude Agent SDK

irm https://claude.ai/set up.ps1 | iex

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

pip set up 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 = """Instructions:
/development   Fast multi-source scan (auto-saves markdown)
/scan    Brief one-page scan
/assist /exit     Assist / Stop
"""

MODEL = os.getenv("CLAUDE_MODEL", "sonnet")  # e.g. "sonnet-4.5"
GRAY = "33[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 "untitled"

def _ensure_dir(p: Path):
    attempt:
        p.mkdir(mother and father=True, exist_ok=True)
    besides Exception:
        go

def _safe_write(path: Path, textual content: str) -> Path:
    """Write textual content to path; if listing/permission fails, fall again to ~/TrendSmith/reviews."""
    attempt:
        _ensure_dir(path.dad or mum)
        path.write_text(textual content, encoding="utf-8")
        return path
    besides Exception:
        home_reports = Path.house() / "TrendSmith"https://www.kdnuggets.com/"reviews"
        _ensure_dir(home_reports)
        fb = home_reports / path.identify
        fb.write_text(textual content, encoding="utf-8")
        return fb

def save_report(subject: str, textual content: str) -> Path:
    filename = f"{_sanitize(subject)}_{_ts()}.md"
    goal = REPORTS / filename
    return _safe_write(goal, textual content.strip() + "n")

class State:
    def __init__(self):
        self.temporary = ""
        self.model_raw = None
        self.utilization = {}
        self.price = None
        self.last_cmd = None
        self.last_topic = None
        self.instruments = {}
        self.t0 = 0.0
        self.t1 = 0.0

    def reset(self):
        self.temporary = ""
        self.model_raw = None
        self.utilization = {}
        self.price = None
        self.instruments = {}
        self.t0 = time.perf_counter()
        self.t1 = 0.0

def friendly_model(identify: str | None) -> str:
    if not identify:
        return MODEL
    n = (identify or "").decrease()
    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 identify or "Unknown"

def usage_footer(st: State, opts_model: str):
    st.t1 = st.t1 or time.perf_counter()
    dur = st.t1 - st.t0
    utilization = st.utilization or {}
    it = utilization.get("input_tokens")
    ot = utilization.get("output_tokens")
    complete = utilization.get("total_tokens")
    if complete is None and (it's not None or ot just isn't None):
        complete = (it or 0) + (ot or 0)
    tools_used = ", ".be a part of(f"{ok}×{v}" for ok, v in st.instruments.objects()) or "--"
    model_label = friendly_model(st.model_raw or opts_model)

    field = [
        "┌─ 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".be a part of(field) + RESET, file=sys.stderr)

async def predominant():
    """Setup → REPL → parse → question/stream → auto-save → abstract."""
    st = State()
    _ensure_dir(REPORTS)

    opts = ClaudeAgentOptions(
        mannequin=MODEL,
        system_prompt=SYS,
        allowed_tools=["WebFetch", "WebSearch"],
    )

    print("📈 TrendSmith nn" + HELP)

    async with ClaudeSDKClient(choices=opts) as consumer:
        whereas True:
            # Learn enter
            attempt:
                person = enter("nYou: ").strip()
            besides (EOFError, KeyboardInterrupt):
                print("nBye!")
                break

            if not person:
                proceed
            low = person.decrease()

            # Fundamental instructions
            if low in {"/exit", "exit", "stop"}:
                print("Bye!")
                break
            if low in {"/assist", "assist"}:
                print(HELP)
                proceed

            # Parse right into a immediate
            if low.startswith("/development "):
                subject = person.cut up(" ", 1)[1].strip().strip('"')
                if not subject:
                    print('e.g. /development "AI chip startups"')
                    proceed
                st.last_cmd, st.last_topic = "development", subject
                immediate = f"Run a quick development scan for '{subject}' following the output spec."
            elif low.startswith("/scan "):
                q = person.cut up(" ", 1)[1].strip()
                if not q:
                    print('e.g. /scan "AI {hardware} information"')
                    proceed
                st.last_cmd, st.last_topic = "scan", q
                immediate = f"Fast scan for '{q}' in below 10s (≤1 WebFetch). Return 5 bullets + TL;DR + sources."
            else:
                st.last_cmd, st.last_topic = "free", None
                immediate = person

            # Execute request and stream outcomes
            st.reset()
            print(f"{GRAY}▶ Working...{RESET}")
            attempt:
                await consumer.question(immediate)
            besides Exception as e:
                print(f"{GRAY}❌ Question error: {e}{RESET}")
                proceed

            attempt:
                async for m in consumer.receive_response():
                    if isinstance(m, AssistantMessage):
                        st.model_raw = st.model_raw or m.mannequin
                        for b in m.content material:
                            if isinstance(b, TextBlock):
                                st.temporary += b.textual content or ""
                                print(b.textual content or "", finish="")
                            elif isinstance(b, ToolUseBlock):
                                identify = b.identify or "Device"
                                st.instruments[name] = st.instruments.get(identify, 0) + 1
                                print(f"{GRAY}n🛠 Device: {identify}{RESET}")
                            elif isinstance(b, ToolResultBlock):
                                go  # quiet software payloads
                    elif isinstance(m, ResultMessage):
                        st.utilization = m.utilization or {}
                        st.price = m.total_cost_usd
            besides Exception as e:
                print(f"{GRAY}n⚠ Stream error: {e}{RESET}")

            # Auto-save development briefs and present the abstract
            if st.last_cmd == "development" and st.temporary.strip():
                attempt:
                    saved_path = save_report(st.last_topic or "development", st.temporary)
                    print(f"n{GRAY}✅ Auto-saved → {saved_path}{RESET}")
                besides Exception as e:
                    print(f"{GRAY}⚠ Save error: {e}{RESET}")

            st.t1 = time.perf_counter()
            usage_footer(st, opts.mannequin)

if __name__ == "__main__":
    asyncio.run(predominant())

/development "AI chip startups"