main() — anthropic-sdk-python Function Reference

Architecture documentation for the main() function in web_search_stream.py from the anthropic-sdk-python codebase.

Function python AnthropicClient SyncAPI called by 1

Entity Profile

AnthropicClient→ SyncAPI→ main() — anthropic-sdk-python Function Reference

Dependency Diagram

graph TD
  3a23204c_8001_a402_6c16_ed2b8c87e0dc["main()"]
  1cb21b00_539d_e8e1_23bf_a71f7566dfa9["web_search_stream.py"]
  3a23204c_8001_a402_6c16_ed2b8c87e0dc -->|defined in| 1cb21b00_539d_e8e1_23bf_a71f7566dfa9
  ce5fcf41_5f71_9717_585d_adf4e8a9b4c5["asyncio()"]
  ce5fcf41_5f71_9717_585d_adf4e8a9b4c5 -->|calls| 3a23204c_8001_a402_6c16_ed2b8c87e0dc
  style 3a23204c_8001_a402_6c16_ed2b8c87e0dc fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

examples/web_search_stream.py lines 6–57

async def main() -> None:
    client = AsyncAnthropic()

    print("Claude with Web Search (Streaming)")
    print("==================================")

    # Create an async stream with web search enabled
    async with client.beta.messages.stream(
        model="claude-sonnet-4-5-20250929",
        max_tokens=1024,
        messages=[{"role": "user", "content": "What's the weather in New York?"}],
        tools=[
            {
                "name": "web_search",
                "type": "web_search_20250305",
            }
        ],
    ) as stream:
        # Process streaming events
        async for chunk in stream:
            # Print text deltas as they arrive
            if chunk.type == "content_block_delta" and chunk.delta.type == "text_delta":
                print(chunk.delta.text, end="", flush=True)

            # Track when web search is being used
            elif chunk.type == "content_block_start" and chunk.content_block.type == "web_search_tool_result":
                print("\n[Web search started...]", end="", flush=True)

            elif chunk.type == "content_block_stop" and chunk.content_block.type == "web_search_tool_result":
                print("[Web search completed]", end="\n\n", flush=True)

        # Get the final complete message
        message = await stream.get_final_message()

    print("\n\nFinal usage statistics:")
    print(f"Input tokens: {message.usage.input_tokens}")
    print(f"Output tokens: {message.usage.output_tokens}")

    if message.usage.server_tool_use:
        print(f"Web search requests: {message.usage.server_tool_use.web_search_requests}")
    else:
        print("No web search requests recorded in usage")

    # Rather than parsing the web search results structure (which varies),
    # we'll just show the complete message structure for debugging
    print("\nMessage Content Types:")
    for i, block in enumerate(message.content):
        print(f"Content Block {i + 1}: Type = {block.type}")

    # Show the entire message structure as JSON for debugging
    print("\nComplete message structure (JSON):")
    print(message.model_dump_json(indent=2))