Erste Schritte mit dem 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:
/pattern   Fast multi-source scan (auto-saves markdown)
/scan    Quick 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 might be TrendSmith, a quick, concise pattern researcher.
- End rapidly (~20 s).
- For /pattern: ≤1 WebSearch + ≤2 WebFetch from distinct domains.
- For /scan: ≤1 WebFetch solely.
Return for /pattern:
 TL;DR (1 line)
 3-5 Alerts (quick bullets)
 Key Gamers, Dangers, 30/90-day Watchlist
 Sources (markdown: **Title** -- URL)
Return for /scan: 5 bullets + TL;DR + Sources.
After ending /pattern, the consumer will auto-save your full transient.
"""

BASE = Path(__file__).father or mother
REPORTS = BASE / "stories"

def _ts():
    return datetime.now().strftime("%Ypercentmpercentd_percentHpercentM")

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:
        move

def _safe_write(path: Path, textual content: str) -> Path:
    """Write textual content to path; if listing/permission fails, fall again to ~/TrendSmith/stories."""
    attempt:
        _ensure_dir(path.father or mother)
        path.write_text(textual content, encoding="utf-8")
        return path
    besides Exception:
        home_reports = Path.house() / "TrendSmith"https://www.kdnuggets.com/"stories"
        _ensure_dir(home_reports)
        fb = home_reports / path.title
        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.transient = ""
        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.transient = ""
        self.model_raw = None
        self.utilization = {}
        self.price = None
        self.instruments = {}
        self.t0 = time.perf_counter()
        self.t1 = 0.0

def friendly_model(title: str | None) -> str:
    if not title:
        return MODEL
    n = (title 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 title 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")
    whole = utilization.get("total_tokens")
    if whole is None and (it's not None or ot just isn't None):
        whole = (it or 0) + (ot or 0)
    tools_used = ", ".be part of(f"{okay}×{v}" for okay, v in st.instruments.objects()) or "--"
    model_label = friendly_model(st.model_raw or opts_model)

    field = (
        "┌─ Run Abstract ─────────────────────────────────────────────",
        f"│ Mannequin: {model_label}",
        f"│ Tokens: {whole if whole just isn't None else '?'}"
        + (f" (in={it if it's not None else '?'} | out={ot if ot just isn't None else '?'})"
            if (it's not None or ot just isn't None) else ""),
        f"│ Instruments: {tools_used}",
        f"│ Period: {dur:.1f}s",
        "└───────────────────────────────────────────────────────────",
    )
    print(GRAY + "n".be part of(field) + RESET, file=sys.stderr)

async def important():
    """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:
                consumer = enter("nYou: ").strip()
            besides (EOFError, KeyboardInterrupt):
                print("nBye!")
                break

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

            # Primary 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("/pattern "):
                subject = consumer.break up(" ", 1)(1).strip().strip('"')
                if not subject:
                    print('e.g. /pattern "AI chip startups"')
                    proceed
                st.last_cmd, st.last_topic = "pattern", subject
                immediate = f"Run a quick pattern scan for '{subject}' following the output spec."
            elif low.startswith("/scan "):
                q = consumer.break 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 beneath 10s (≤1 WebFetch). Return 5 bullets + TL;DR + sources."
            else:
                st.last_cmd, st.last_topic = "free", None
                immediate = consumer

            # 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.transient += b.textual content or ""
                                print(b.textual content or "", finish="")
                            elif isinstance(b, ToolUseBlock):
                                title = b.title or "Instrument"
                                st.instruments(title) = st.instruments.get(title, 0) + 1
                                print(f"{GRAY}n🛠 Instrument: {title}{RESET}")
                            elif isinstance(b, ToolResultBlock):
                                move  # 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 pattern briefs and present the abstract
            if st.last_cmd == "pattern" and st.transient.strip():
                attempt:
                    saved_path = save_report(st.last_topic or "pattern", st.transient)
                    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(important())

/pattern "AI chip startups"