import anthropic

client = anthropic.Anthropic()

def stream_basic_response(user_message: str):
    """Stream a basic Claude response token by token."""
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # Newline after stream completes

stream_basic_response("Explain quantum entanglement in simple terms.")

import anthropic
import asyncio

async_client = anthropic.AsyncAnthropic()

async def stream_basic_response_async(user_message: str):
    """Async streaming with the Anthropic SDK."""
    async with async_client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
    print()

asyncio.run(stream_basic_response_async("Explain quantum entanglement in simple terms."))

import anthropic
import json

client = anthropic.Anthropic()

# Define tools
tools = [
    {
        "name": "search_orders",
        "description": "Search customer orders by order ID or customer email.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Order ID or email"},
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_shipping_status",
        "description": "Get real-time shipping status for an order.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "The order ID"},
            },
            "required": ["order_id"],
        },
    },
]


def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool and return the result as a string."""
    if tool_name == "search_orders":
        # Simulated database lookup
        return json.dumps({
            "order_id": "ORD-12345",
            "customer": "jane@example.com",
            "items": ["Blue Widget x2", "Red Gadget x1"],
            "total": "$47.99",
            "status": "shipped",
        })
    elif tool_name == "get_shipping_status":
        return json.dumps({
            "order_id": tool_input["order_id"],
            "carrier": "FedEx",
            "tracking": "7891011",
            "estimated_delivery": "2026-03-10",
            "current_location": "Memphis, TN",
        })
    return json.dumps({"error": f"Unknown tool: {tool_name}"})


def stream_agent_response(user_message: str):
    """
    Complete streaming agent loop with real-time tool call display.
    """
    messages = [{"role": "user", "content": user_message}]

    while True:
        # Stream the model response
        tool_calls = []
        current_tool = None

        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        ) as stream:
            response = None
            for event in stream:
                # The SDK exposes raw events via iteration
                pass

            # Use the helper to collect text and tool use
            response = stream.get_final_message()

        # Process the response content blocks
        for block in response.content:
            if block.type == "text":
                print(block.text, end="", flush=True)
            elif block.type == "tool_use":
                tool_name = block.name
                tool_input = block.input
                tool_id = block.id

                # Show the user what's happening
                print(f"\n⚙️  Calling tool: {tool_name}({json.dumps(tool_input)})")

                # Execute the tool
                result = execute_tool(tool_name, tool_input)
                print(f"✅  Result received from {tool_name}")

                tool_calls.append({
                    "tool_use_id": tool_id,
                    "tool_name": tool_name,
                    "tool_input": tool_input,
                    "result": result,
                })

        # If the model stopped because it wants to use tools, continue the loop
        if response.stop_reason == "tool_use":
            # Add assistant message with all content blocks
            messages.append({"role": "assistant", "content": response.content})

            # Add tool results
            tool_results = []
            for tc in tool_calls:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tc["tool_use_id"],
                    "content": tc["result"],
                })
            messages.append({"role": "user", "content": tool_results})

            print("\n--- Continuing agent loop ---")
        else:
            # Model is done (stop_reason == "end_turn")
            print()
            break


# Run the agent
stream_agent_response("Where is my order #ORD-12345? When will it arrive?")

def stream_agent_with_live_tokens(user_message: str):
    """Stream text tokens live while also detecting tool calls."""
    messages = [{"role": "user", "content": user_message}]

---

## Verwandte Artikel

- [Tool-Nutzungsmuster: Zuverlässige Agent-Tool-Schnittstellen entwickeln](/de/blog/agent-tool-use-patterns/)
- [Multi-Agenten-Muster: Orchestratoren, Worker und Pipelines](/de/blog/multi-agent-patterns/)
- [Agent-Fehlerbehandlung: 5 Muster für Produktionszuverlässigkeit](/de/blog/agent-error-recovery-patterns/)
- [Web-Automatisierungsagenten: Browsersteuerung mit Claude und Computer Use](/de/blog/web-automation-agents-browser-control-with-claude-and-computer-use/)