1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import asyncio
from anthropic import AsyncAnthropic
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))
if __name__ == "__main__":
asyncio.run(main())
|
