claude-agent-sdk

name	claude-agent-sdk
description	Integrate Claude Agent SDK into Python apps with subscription OAuth, structured output, MCP tools, and provider abstraction. Captures production-validated patterns and bug workarounds. Use when adding Claude Agent SDK to a Python application, enabling Claude subscription auth (OAuth) alongside API key auth, building multi-turn agent workflows with tool use, integrating MCP tools, or setting up provider abstraction.
version	1.0.0
created	"2026-04-24T00:00:00.000Z"
updated	"2026-04-24T00:00:00.000Z"
tags	["claude","agent-sdk","anthropic","oauth","subscription","llm","mcp"]
spec_ref	.claude/skills/claude-agent-sdk/SPEC.md

Claude Agent SDK Integration

Production-validated patterns from MeatyWiki (90+ SP across three feature tracks: LLM Flexibility, Agent SDK MVP, Subscription Auth). Full architecture contract in SPEC.md.

When to Use

• Adding Claude Agent SDK (anthropic package) to a Python application
• Enabling subscription OAuth auth alongside API key auth
• Building multi-turn agent workflows with tool use
• Integrating MCP tools for read-only data access
• Setting up provider abstraction (agent mode + local LLMs + API providers)
• Debugging structured_output: None or 404 errors under subscription auth

Architecture Overview

Application Stage
    ├── Tool-free / simple     → LLMClient.chat()      (sync, LiteLLM)
    └── Tool-using / multi-turn → AgentRuntime.run()   (async, Agent SDK)
                                    ├→ AgentBridge      (sync/async wrapper)
                                    └→ MCP Server       (read-only tools, subprocess)

Opt-in per-purpose. Configs without an agent: key load unchanged (backward compat).

---

Quick Start: Minimal Integration

# 1. Define per-purpose agent config
class AgentConfig(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")
    system_prompt: str
    tools: list[str] = Field(default_factory=list)
    max_turns: int = Field(default=6, ge=1, le=10)
    output_schema: dict | None = None

# 2. Route at call time
def process(purpose: str, messages: list[dict]) -> str:
    model_cfg = get_model_config(purpose)
    if model_cfg.agent is not None:
        return agent_run(purpose, messages)   # async Agent SDK path
    else:
        return chat(purpose, messages)        # sync LiteLLM path

---

Core Pattern: Sync/Async Bridge

The Agent SDK is async. Wrap it for sync callers. Overhead: 40–90 µs per call.

import anyio

class AgentBridge:
    """Wraps async Agent SDK in sync context."""

    def __init__(self):
        self._portal = None

    def _ensure_portal(self):
        if self._portal is None:
            self._portal = anyio.from_thread.start_blocking_portal()

    def run_sync(self, coro):
        self._ensure_portal()
        return self._portal.call(coro)

    def close(self):
        if self._portal is not None:
            self._portal.call(self._portal.stop)
            self._portal = None


class FakeAgentBridge:
    """Test double — runs on fresh asyncio loop. No pytest-asyncio needed."""

    def run_sync(self, coro):
        import asyncio
        return asyncio.run(coro)

    def close(self):
        pass

Key facts:

• anyio.from_thread.start_blocking_portal() creates a thread-pinned event loop
• Works in plain sync, pytest, and arq worker contexts
• Register atexit handler calling bridge.close() to avoid dangling threads
• close() must be idempotent

---

Core Pattern: Agent Runtime

class AgentRuntime:
    def __init__(
        self,
        config: AgentConfig,
        provider: str,
        model: str,
        db_path: Path | None = None,
        purpose: str | None = None,
    ):
        self._purpose = purpose or "unknown"   # MUST store — used in turn events
        self._config = config
        self._model = model
        self._provider = provider

        # Auto-populate output_schema from registry if not provided
        if config.output_schema is None and purpose:
            schema = SCHEMA_REGISTRY.get(purpose)
            if schema is not None:
                self._config = config.model_copy(update={"output_schema": schema})

        # Validate: output_schema requires max_turns >= 2
        if self._config.output_schema and self._config.max_turns < 2:
            raise ValueError(
                f"max_turns >= 2 required when output_schema is set "
                f"(purpose={self._purpose}, got max_turns={self._config.max_turns})"
            )

    @property
    def has_schema(self) -> bool:
        return self._config.output_schema is not None

    async def run(self, messages: list[dict]) -> dict:
        run_id = ulid.new().str
        options = {
            "model": self._model,
            "max_turns": self._config.max_turns,
            "system_prompt": self._config.system_prompt,
        }
        if self._config.output_schema:
            options["output_format"] = {
                "type": "json_schema",
                "schema": self._config.output_schema,
            }
        if self._config.tools:
            options["mcp_servers"] = [self._build_mcp_server()]

        # Drive the SDK turn loop, emit llm.agent.turn per turn
        # Return: {text, structured_output, cost_estimate, turns, tokens_in, tokens_out}
        ...

---

Core Pattern: Result Adaptation + Fallback

def agent_run(self, purpose: str, messages: list[dict]) -> LLMCallResult:
    try:
        runtime = self.router.get_agent_runtime(purpose)
        sdk_result = self.bridge.run_sync(runtime.run(messages))

        # CRITICAL: sdk_result may be dict or object depending on SDK version
        raw_output = _field(sdk_result, "structured_output", None)

        # Text-parse fallback for subscription auth (JSON returned as plain text)
        if raw_output is None and runtime.has_schema:
            text = _field(sdk_result, "text", "")
            raw_output = _try_parse_json_payload(text)
            source = "text_parse"
        else:
            source = "sdk"

        return LLMCallResult(
            content=raw_output or _field(sdk_result, "text", ""),
            cost_estimate=_field(sdk_result, "cost_estimate", 0.0),
            structured_output_source=source,
        )
    except AgentRuntimeError as e:
        log_event("llm.agent.run_failed", error=str(e))
        # Fallback to sync — system prompt EXCLUDED (it may reference unavailable tools)
        result = self.chat(purpose, [m for m in messages if m["role"] == "user"])
        result.degraded = True
        return result


def _field(obj: Any, key: str, default: Any = None) -> Any:
    """Read from dict or object — handles both SDK result shapes."""
    if isinstance(obj, dict):
        return obj.get(key, default)
    return getattr(obj, key, default)


def _try_parse_json_payload(text: str) -> dict | None:
    """Three recovery strategies for subscription auth structured output."""
    import json, re

    # Strategy 1: Direct parse
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    # Strategy 2: Markdown code-fence extraction
    match = re.search(r"```(?:json)?\s*\n(.*?)\n```", text, re.DOTALL)
    if match:
        try:
            return json.loads(match.group(1))
        except json.JSONDecodeError:
            pass

    # Strategy 3: Brace/bracket balancing for embedded JSON
    for start_char, end_char in [("{", "}"), ("[", "]")]:
        idx = text.find(start_char)
        if idx >= 0:
            depth, end = 0, idx
            for i, c in enumerate(text[idx:], idx):
                if c == start_char:
                    depth += 1
                elif c == end_char:
                    depth -= 1
                if depth == 0:
                    end = i + 1
                    break
            try:
                return json.loads(text[idx:end])
            except json.JSONDecodeError:
                pass

    return None

---

Auth Setup

Interactive / development:

claude /login
# Stores in macOS Keychain. Auto-refreshes on 401. ~6–8 hr lifetime.

Daemon / headless / workers (production):

claude setup-token
# One-time interactive step → one-year token
export CLAUDE_CODE_OAUTH_TOKEN="<token>"
# arq/cron workers inherit this at job dispatch time

API key fallback:

export ANTHROPIC_API_KEY="sk-ant-..."

Resolution order: Keychain → CLAUDE_CODE_OAUTH_TOKEN → ANTHROPIC_API_KEY

Critical: Agent SDK subprocess is the sanctioned path for subscription usage. Routing subscription OAuth tokens through LiteLLM directly is a ToS grey area for non-interactive API calls. Always use the Agent SDK path for subscription auth.

Linux headless: Keychain unavailable. CLAUDE_CODE_OAUTH_TOKEN is the only viable path.

---

Provider Abstraction

class ProviderCapabilities(BaseModel):
    agent_mode: Literal["native", "proxy", "none"] = "none"

class ProviderDescriptor(BaseModel):
    adapter: str = "openai_compat"
    base_url: str | None = None
    api_key_env: str | None = None
    extra_headers: dict[str, str] = Field(default_factory=dict)
    cost_per_1m_tokens: dict[str, float] = Field(default_factory=dict)
    capabilities: ProviderCapabilities = Field(default_factory=ProviderCapabilities)

Validate at router init — reject agent: config when provider agent_mode != "native":

if stage_cfg.agent is not None and provider.capabilities.agent_mode != "native":
    raise ConfigError(
        f"Stage '{purpose}' has agent: config but provider '{provider_id}' "
        f"has agent_mode={provider.capabilities.agent_mode!r} (need 'native')"
    )

Example YAML:

llm:
  providers:
    anthropic:
      adapter: openai_compat
      api_key_env: CLAUDE_CODE_OAUTH_TOKEN
      capabilities:
        agent_mode: native
    ollama:
      adapter: openai_compat
      base_url: http://localhost:11434/v1
      capabilities:
        agent_mode: none   # agent: config rejected at init

models:
  classify:
    provider: anthropic
    model: claude-haiku-4-5-20251001
    agent:
      system_prompt: "You are a document classifier."
      tools: []
      max_turns: 2      # minimum when output_schema is set

  query:
    provider: anthropic
    model: claude-opus-4-6
    agent:
      system_prompt: "You are a vault query assistant."
      tools: [vault_search, vault_read]
      max_turns: 6

---

Schema Registry

# Static dict: purpose → JSON Schema derived from Pydantic models
SCHEMA_REGISTRY: dict[str, dict | None] = {
    "classify": Classification.model_json_schema(),
    "extract":  ExtractedPayload.model_json_schema(),
    "compile":  SummaryContent.model_json_schema(),
    "lint":     ContradictionFinding.schema,   # dataclass: hand-crafted schema
    "query":    None,                          # plain text, no schema
}

Auto-populate AgentConfig.output_schema from this registry at router init. Snapshot tests catch Pydantic model drift silently breaking schemas.

Turn count rule: max_turns >= 2 is required when output_schema is set. The SDK's StructuredOutput tool needs a second turn to return validated output. max_turns=1 with schema → error_max_turns.

---

MCP Tool Integration

# Subprocess-executable MCP server for read-only data access
# Safety guarantees:
#   1. Read-only connection (SQLite: ?mode=ro, or equivalent)
#   2. Per-tool timeout (2 seconds default)
#   3. Payload cap (100 KB, truncate with sentinel)
#   4. Subprocess isolation — no import guards needed

def _build_mcp_server(self) -> dict:
    return {
        "name": "vault",
        "command": sys.executable,
        "args": ["-m", "meatywiki.llm.agent.mcp_server"],
        "env": {
            "VAULT_DB_PATH": str(self._db_path),
            "MCP_READ_ONLY": "1",
        },
    }

---

Observability

Emit per-turn events with ULID correlation. Never log tool content — names only.

log_event("llm.agent.turn", {
    "run_id": run_id,          # ULID, correlates all turns in a run
    "purpose": self._purpose,  # store in __init__ or shows "unknown"
    "provider": self._provider,
    "model": self._model,
    "turn_index": i,
    "tokens_in": tokens_in,
    "tokens_out": tokens_out,
    "cost": cost,
    "latency_ms": latency,
    "stop_reason": stop,
    "tools_invoked": tool_names,   # names ONLY — never content
})

Cost tracking:

• Agent path: ResultMessage.total_cost_usd — SDK-reported, actual billing, most accurate
• Sync fallback: LiteLLM pricing table + per-provider YAML cost_per_1m_tokens overrides
• Both surface through the same cost_estimate field for aggregation

---

Known Bugs and Workarounds

Bug 1 — LiteLLM double /v1/ path (404s)

Affects LiteLLM 1.83+. LiteLLM auto-appends /v1/messages to api_base. If base_url ends with /v1/, result is .../v1//v1/messages.

import re

# Strip trailing /v1/ before passing to LiteLLM
if provider_id == "anthropic" and base_url:
    base_url = re.sub(r"/v1/?$", "", base_url)

Bug 2 — structured_output always None

Two causes:

Cause A: AgentRuntime.run() returns dict, not an object. getattr(dict, "key") returns None silently.

Fix: always use _field(obj, key, default) (shown above).

Cause B: Claude CLI under subscription auth emits JSON as plain text, not in the structured_output field.

Fix: _try_parse_json_payload() three-stage fallback (shown above).

Bug 3 — Purpose shows "unknown" in turn events

Cause: _purpose attribute never stored on the runtime instance.

# In AgentRuntime.__init__:
self._purpose = purpose.value if hasattr(purpose, "value") else (purpose or "unknown")

---

Testing Strategy

class FakeAgentBridge:
    """Sync test double. No pytest-asyncio required."""

    def __init__(self, response: str, structured: dict | None = None):
        self._response = response
        self._structured = structured

    def run_sync(self, coro):
        import asyncio
        # Run real coroutine if needed, or return pre-baked result
        return {
            "text": self._response,
            "structured_output": self._structured,
            "cost_estimate": 0.0,
            "turns": 1,
            "tokens_in": 10,
            "tokens_out": 20,
        }

    def close(self):
        pass

Test coverage checklist:

• Config backward compat — configs without agent: key load unchanged
• Schema registry snapshots — catch Pydantic model drift at CI time
• Both paths per purpose — agent dispatch AND sync fallback for every stage
• Integration tests — patched bridge + real app code paths (no live API call)
• Manual smoke tests — real data for agent mode validation before merging

---

Implementation Checklist

• Add anthropic as optional dependency (e.g., [agent] extra in pyproject.toml)
• Create AgentConfig with per-purpose agent: opt-in key
• Add ProviderCapabilities.agent_mode flag to ProviderDescriptor
• Validate at router init: reject agent: config when agent_mode != "native"
• Implement AgentBridge with anyio.from_thread.start_blocking_portal(); register atexit
• Implement AgentRuntime; store self._purpose in __init__
• Build schema registry; auto-populate output_schema at router init
• Add _field(obj, key, default) helper for dict/object result access
• Add _try_parse_json_payload() three-stage fallback
• Fallback-to-sync on AgentRuntimeError; set degraded=True; exclude system prompt
• Strip trailing /v1/ from Anthropic base_url before LiteLLM init
• Enforce max_turns >= 2 when output_schema is set
• Emit per-turn events with ULID run_id
• Document daemon auth setup (claude setup-token)
• Add FakeAgentBridge test double
• Snapshot tests: config compat + schema registry drift
• MCP tools (if needed): subprocess, read-only FS, per-tool timeout, payload cap

---

Common Mistakes

Mistake	Why It Fails	Fix
max_turns=1 with output_schema	StructuredOutput tool needs a second turn	Set max_turns >= 2
getattr(dict_result, "key")	Dict doesn't support getattr	Use _field() helper
System prompt in fallback	May reference tools unavailable in sync path	Exclude system prompt from fallback call
LiteLLM with subscription OAuth	ToS grey area for non-interactive calls	Use Agent SDK subprocess path
Not storing _purpose on runtime	Turn events all show purpose: "unknown"	self._purpose = purpose or "unknown" in __init__
Keychain token for daemon/cron	6–8 hr expiry; no refresh in background	Use CLAUDE_CODE_OAUTH_TOKEN
api_base ending with /v1/	LiteLLM appends /v1/messages → double path	Strip trailing /v1/? before LiteLLM init

---

OAuth Routing Pitfalls

Two approaches that appear plausible but do not work with subscription OAuth tokens (sk-ant-oat01-...).

Pitfall 1 — Routing OAuth tokens via Anthropic(auth_token=...)

# DOES NOT WORK
import anthropic
client = anthropic.Anthropic(auth_token=os.environ["CLAUDE_CODE_OAUTH_TOKEN"])
client.messages.create(...)
# → 401 "OAuth authentication is currently not supported."

The Anthropic Messages API rejects subscription OAuth tokens at the HTTP layer. The standard anthropic Python SDK has no supported path for these tokens — the auth_token parameter routes to the same bearer-token endpoint that the API actively rejects for subscription credentials.

Fix: Use claude_agent_sdk.query() dispatched through AgentBridge.run_sync(). The agent SDK is a thin async wrapper around the claude CLI subprocess; the CLI handles subscription OAuth via the CLAUDE_CODE_OAUTH_TOKEN env var (forwarded by AgentBridge.build_subprocess_env() and ClaudeAgentOptions(env=...)) and is the only sanctioned execution path for subscription auth.

# Works — agent SDK subprocess path via AgentBridge
from claude_agent_sdk import ClaudeAgentOptions, query

async def _run():
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-6",
        max_turns=1,
        thinking={"type": "disabled"},
        env=bridge.build_subprocess_env(),  # injects CLAUDE_CODE_OAUTH_TOKEN
        setting_sources=[],
    )
    text = []
    async for msg in query(prompt=prompt, options=options):
        # accumulate AssistantMessage TextBlock content...
        ...
    return "".join(text)

result_text = bridge.run_sync(_run())

Reference implementation: skillmeat/core/discovery/subscription_enrichment_adapter.py (Tier 2 enrichment), skillmeat/core/discovery/agent_runner.py (Tier 3 multi-turn).

Pitfall 2 — Setting anthropic-beta: oauth-2025-04-20 on standard SDK calls

# DOES NOT WORK (and is a ToS grey area)
client = anthropic.Anthropic(
    api_key=os.environ["CLAUDE_CODE_OAUTH_TOKEN"],
    default_headers={"anthropic-beta": "oauth-2025-04-20"},
)

This beta header is scoped to first-party Anthropic apps with specific OAuth grant scopes. It is not a general-purpose enablement flag — applying it in arbitrary third-party callers is a ToS grey area and does not resolve the underlying rejection.

Fix: Same as Pitfall 1 — use the claude CLI subprocess. It manages the OAuth flow without requiring manual header injection.

Attempted Fix	Result	Correct Path
Anthropic(auth_token=oauth_token)	401 — not supported	AgentBridge → CLI subprocess
anthropic-beta: oauth-2025-04-20 header	ToS grey area, unreliable	AgentBridge → CLI subprocess
CLAUDE_CODE_OAUTH_TOKEN + CLI subprocess	Works, sanctioned	Use this

---

Version History

• 2026-04-24: v1.0.0 — Initial skill from MeatyWiki production learnings (90+ SP across LLM Flexibility, Agent SDK MVP, Subscription Auth feature tracks)