Back to modules
Module 5

Advanced SDK Patterns

Hooks, streaming, cost tracking, structured outputs, and production patterns.

📖 35 min
HooksStreamingCost TrackingStructured Outputs

Streaming: Real-Time Output

By default, the SDK collects all messages and returns them. You can stream output in real-time.

Streaming Text

python
from claudeagentsdk import query



async for message in query(


    prompt="Explain agents in 100 words",


    options={"stream": True}


):


    if hasattr(message, "delta"):


        # Partial text update


        print(message.delta, end="", flush=True)




Output streams character by character

Streaming Tool Calls

python
async for message in query(


    prompt="Read file.txt and analyze it",


    options={"stream": True}


):


    if isinstance(message, AssistantMessage):


        for block in message.content:


            if isinstance(block, ToolUseBlock):


                print(f"🔧 Calling {block.name}...")


    


    elif isinstance(message, UserMessage):


        # Tool results coming back


        print(f"📦 Got results")

---

Cost Tracking & Budgeting

Track Costs

python
totalcost = 0



async for message in query(prompt="..."):


    if isinstance(message, ResultMessage):


        cost = message.cost


        print(f"Input tokens: ${cost['input']:.6f}")


        print(f"Output tokens: ${cost['output']:.6f}")


        print(f"Total: ${cost['input'] + cost['output']:.6f}")

Set Budget Limit

python
options = ClaudeAgentOptions(


    maxbudgetusd=0.50  # Stop if cost exceeds 50 cents

)


async for message in query(prompt="...", options=options):


    if isinstance(message, ResultMessage):


        if message.subtype == "errormaxbudget":


            print("Hit budget limit!")

Log All Costs

python
costs = []



async def trackcost(hook: AfterToolResult) -> None:


    costs.append({


        "tool": hook.tooluse.name,


        "cost": hook.cost,


        "tokens": hook.tokens


    })



options = ClaudeAgentOptions(


    hooks={"aftertoolresult": trackcost}

)


async for message in query(prompt="...", options=options):


    pass



print(f"Total cost: ${sum(c['cost'] for c in costs):.4f}")

---

Structured Outputs

Make agents return data in a specific format (JSON schema).

Define Schema

python
from pydantic import BaseModel



class AnalysisResult(BaseModel):


    filename: str


    issuesfound: int


    severity: str  # "critical" | "high" | "medium" | "low"


    fixesapplied: list[str]

Use with Agent

python
from claudeagentsdk import query



async for message in query(


    prompt="Analyze auth.ts for security issues",


    options={


        "outputschema": AnalysisResult,


        "outputjson": True  # Return JSON, not text


    }


):


    if isinstance(message, ResultMessage):


        result = AnalysisResult.modelvalidatejson(message.result)


        print(f"Issues: {result.issuesfound}")


        print(f"Severity: {result.severity}")

Why Structured Outputs?

  • Parsing: Easy to read programmatically

  • Validation: Schema enforces correct format
  • Integration: Feed results into other systems
  • Reliability: No "oops I forgot a field" problems

    • ---

    Retry Logic

    Retry on Failure

    python
    async def runagentwithretry(prompt, maxretries=3):
    

        for attempt in range(maxretries):
    

            try:
    

                async for message in query(prompt=prompt):
    

                    if isinstance(message, ResultMessage):
    

                        if message.subtype == "success":
    

                            return message.result
    

                        # Fall through to retry
    

            except Exception as e:
    

                print(f"Attempt {attempt + 1} failed: {e}")
    

        
    

        raise RuntimeError("Agent failed after all retries")
    

    
result = await runagentwithretry("Read config.json")

    Retry with Backoff

    python
    import asyncio
    

    
async def runwithbackoff(prompt, maxretries=3):
    

        for attempt in range(maxretries):
    

            try:
    

                async for message in query(prompt=prompt):
    

                    if isinstance(message, ResultMessage):
    

                        if message.subtype == "success":
    

                            return message.result
    

            except Exception as e:
    

                if attempt < maxretries - 1:
    

                    waittime = (2  attempt)  # Exponential backoff
    

                    print(f"Retry in {waittime}s...")
    

                    await asyncio.sleep(waittime)
    

                else:
    

                    raise
    

    
result = await runwithbackoff("Complex analysis task")

    ---

    Production Patterns

    Pattern 1: Queue-Based Processing

    python
    import asyncio
    

    from queue import Queue
    

    
taskqueue = Queue()
    

    
async def processtasks():
    

        while True:
    

            if taskqueue.empty():
    

                await asyncio.sleep(1)
    

                continue
    

            
    

            task = taskqueue.get()
    

            print(f"Processing: {task}")
    

            
    

            async for message in query(prompt=f"Process: {task}"):
    

                if isinstance(message, ResultMessage):
    

                    print(f"✓ {task} completed")
    

                    taskqueue.taskdone()
    

    

    Add tasks
    

    taskqueue.put("Audit auth.ts")
    

    taskqueue.put("Test API endpoints")
    

    taskqueue.put("Review database schema")
    

    
asyncio.run(processtasks())

    Pattern 2: Rate Limiting

    python
    import asyncio
    

    from datetime import datetime, timedelta
    

    
lastcall = None
    

    mininterval = 1  # Min 1 second between calls
    

    
async def ratelimitedquery(prompt):
    

        global lastcall
    

        
    

        if lastcall:
    

            elapsed = (datetime.now() - lastcall).totalseconds()
    

            if elapsed < mininterval:
    

                await asyncio.sleep(mininterval - elapsed)
    

        
    

        lastcall = datetime.now()
    

        
    

        async for message in query(prompt=prompt):
    

            yield message

    Pattern 3: Caching Results

    python
    from pathlib import Path
    

    import json
    

    
cachedir = Path(".agentcache")
    

    cachedir.mkdir(existok=True)
    

    
async def cachedquery(prompt):
    

        cachefile = cachedir / f"{hash(prompt)}.json"
    

        
    

        # Check cache
    

        if cachefile.exists():
    

            with open(cachefile) as f:
    

                return json.load(f)["result"]
    

        
    

        # Run agent
    

        result = None
    

        async for message in query(prompt=prompt):
    

            if isinstance(message, ResultMessage):
    

                result = message.result
    

        
    

        # Save to cache
    

        with open(cache_file, "w") as f:
    

            json.dump({"result": result}, f)
    

        
    

        return result

    ---

    Key Gotchas

    Gotcha #1: Streaming changes message flow

  • With streaming, messages arrive incrementally
  • Need to handle partial messages
  • Different from non-streaming patterns

    • Gotcha #2: Hooks execute on main thread

  • Don't block for long in hooks
  • Use async hooks for I/O operations
  • Blocking = delays entire agent loop

    • Gotcha #3: Structured outputs add cost

  • Schema validation uses tokens
  • More complex schemas = higher cost
  • Sometimes plain text is cheaper

    • Gotcha #4: Retries compound costs

  • Each retry = full agent run
  • Budget limits still apply per retry
  • Consider if retries are worth it

    • ---

    Next Steps

    Now you understand advanced SDK patterns. Next module: Subagents in the SDK** — creating specialized agents that work together!

    Previous
    Tools & Tool Use
    Next
    Subagents in the SDK