skills/claude-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: claude-api skill

Browse Source
This commit is contained in:
zlei9
2026-03-22 11:41:50 +08:00
commit 268400a2dc
26 changed files with 5454 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

269
python/agent-sdk/README.md Normal file
View File

@@ -0,0 +1,269 @@
# Agent SDK — Python
The Claude Agent SDK provides a higher-level interface for building AI agents with built-in tools, safety features, and agentic capabilities.
## Installation
```bash
pip install claude-agent-sdk
```
---
## Quick Start
```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async def main():
async for message in query(
prompt="Explain this codebase",
options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])
):
if isinstance(message, ResultMessage):
print(message.result)
anyio.run(main)
```
---
## Built-in Tools
| Tool | Description |
| --------- | ------------------------------------ |
| Read | Read files in the workspace |
| Write | Create new files |
| Edit | Make precise edits to existing files |
| Bash | Execute shell commands |
| Glob | Find files by pattern |
| Grep | Search files by content |
| WebSearch | Search the web for information |
| WebFetch | Fetch and analyze web pages |
| AskUserQuestion | Ask user clarifying questions |
| Agent | Spawn subagents |
---
## Primary Interfaces
### `query()` — Simple One-Shot Usage
The `query()` function is the simplest way to run an agent. It returns an async iterator of messages.
```python
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async for message in query(
prompt="Explain this codebase",
options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])
):
if isinstance(message, ResultMessage):
print(message.result)
```
### `ClaudeSDKClient` — Full Control
`ClaudeSDKClient` provides full control over the agent lifecycle. Use it when you need custom tools, hooks, streaming, or the ability to interrupt execution.
```python
import anyio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock
async def main():
options = ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])
async with ClaudeSDKClient(options=options) as client:
await client.query("Explain this codebase")
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
anyio.run(main)
```
`ClaudeSDKClient` supports:
- **Context manager** (`async with`) for automatic resource cleanup
- **`client.query(prompt)`** to send a prompt to the agent
- **`receive_response()`** for streaming messages until completion
- **`interrupt()`** to stop agent execution mid-task
- **Required for custom tools** (via SDK MCP servers)
---
## Permission System
```python
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async for message in query(
prompt="Refactor the authentication module",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Edit", "Write"],
permission_mode="acceptEdits" # Auto-accept file edits
)
):
if isinstance(message, ResultMessage):
print(message.result)
```
Permission modes:
- `"default"`: Prompt for dangerous operations
- `"plan"`: Planning only, no execution
- `"acceptEdits"`: Auto-accept file edits
- `"dontAsk"`: Don't prompt (useful for CI/CD)
- `"bypassPermissions"`: Skip all prompts (requires `allow_dangerously_skip_permissions=True` in options)
---
## MCP (Model Context Protocol) Support
```python
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async for message in query(
prompt="Open example.com and describe what you see",
options=ClaudeAgentOptions(
mcp_servers={
"playwright": {"command": "npx", "args": ["@playwright/mcp@latest"]}
}
)
):
if isinstance(message, ResultMessage):
print(message.result)
```
---
## Hooks
Customize agent behavior with hooks using callback functions:
```python
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher, ResultMessage
async def log_file_change(input_data, tool_use_id, context):
file_path = input_data.get('tool_input', {}).get('file_path', 'unknown')
print(f"Modified: {file_path}")
return {}
async for message in query(
prompt="Refactor utils.py",
options=ClaudeAgentOptions(
permission_mode="acceptEdits",
hooks={
"PostToolUse": [HookMatcher(matcher="Edit|Write", hooks=[log_file_change])]
}
)
):
if isinstance(message, ResultMessage):
print(message.result)
```
Available hook events: `PreToolUse`, `PostToolUse`, `PostToolUseFailure`, `Notification`, `UserPromptSubmit`, `SessionStart`, `SessionEnd`, `Stop`, `SubagentStart`, `SubagentStop`, `PreCompact`, `PermissionRequest`, `Setup`, `TeammateIdle`, `TaskCompleted`, `ConfigChange`
---
## Common Options
`query()` takes a top-level `prompt` (string) and an `options` object (`ClaudeAgentOptions`):
```python
async for message in query(prompt="...", options=ClaudeAgentOptions(...)):
```
| Option | Type | Description |
| ----------------------------------- | ------ | -------------------------------------------------------------------------- |
| `cwd` | string | Working directory for file operations |
| `allowed_tools` | list | Tools the agent can use (e.g., `["Read", "Edit", "Bash"]`) |
| `tools` | list | Built-in tools to make available (restricts the default set) |
| `disallowed_tools` | list | Tools to explicitly disallow |
| `permission_mode` | string | How to handle permission prompts |
| `allow_dangerously_skip_permissions`| bool | Must be `True` to use `permission_mode="bypassPermissions"` |
| `mcp_servers` | dict | MCP servers to connect to |
| `hooks` | dict | Hooks for customizing behavior |
| `system_prompt` | string | Custom system prompt |
| `max_turns` | int | Maximum agent turns before stopping |
| `max_budget_usd` | float | Maximum budget in USD for the query |
| `model` | string | Model ID (default: determined by CLI) |
| `agents` | dict | Subagent definitions (`dict[str, AgentDefinition]`) |
| `output_format` | dict | Structured output schema |
| `thinking` | dict | Thinking/reasoning control |
| `betas` | list | Beta features to enable (e.g., `["context-1m-2025-08-07"]`) |
| `setting_sources` | list | Settings to load (e.g., `["project"]`). Default: none (no CLAUDE.md files) |
| `env` | dict | Environment variables to set for the session |
---
## Message Types
```python
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage, SystemMessage
async for message in query(
prompt="Find TODO comments",
options=ClaudeAgentOptions(allowed_tools=["Read", "Glob", "Grep"])
):
if isinstance(message, ResultMessage):
print(message.result)
elif isinstance(message, SystemMessage) and message.subtype == "init":
session_id = message.session_id # Capture for resuming later
```
---
## Subagents
```python
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition, ResultMessage
async for message in query(
prompt="Use the code-reviewer agent to review this codebase",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep", "Agent"],
agents={
"code-reviewer": AgentDefinition(
description="Expert code reviewer for quality and security reviews.",
prompt="Analyze code quality and suggest improvements.",
tools=["Read", "Glob", "Grep"]
)
}
)
):
if isinstance(message, ResultMessage):
print(message.result)
```
---
## Error Handling
```python
from claude_agent_sdk import query, ClaudeAgentOptions, CLINotFoundError, CLIConnectionError, ResultMessage
try:
async for message in query(
prompt="...",
options=ClaudeAgentOptions(allowed_tools=["Read"])
):
if isinstance(message, ResultMessage):
print(message.result)
except CLINotFoundError:
print("Claude Code CLI not found. Install with: pip install claude-agent-sdk")
except CLIConnectionError as e:
print(f"Connection error: {e}")
```
---
## Best Practices
1. **Always specify allowed_tools** — Explicitly list which tools the agent can use
2. **Set working directory** — Always specify `cwd` for file operations
3. **Use appropriate permission modes** — Start with `"default"` and only escalate when needed
4. **Handle all message types** — Check for `ResultMessage` to get agent output
5. **Limit max_turns** — Prevent runaway agents with reasonable limits

319
python/agent-sdk/patterns.md Normal file
View File

@@ -0,0 +1,319 @@
# Agent SDK Patterns — Python
## Basic Agent
```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async def main():
async for message in query(
prompt="Explain what this repository does",
options=ClaudeAgentOptions(
cwd="/path/to/project",
allowed_tools=["Read", "Glob", "Grep"]
)
):
if isinstance(message, ResultMessage):
print(message.result)
anyio.run(main)
```
---
## Custom Tools
Custom tools require an MCP server. Use `ClaudeSDKClient` for full control, or pass the server to `query()` via `mcp_servers`.
```python
import anyio
from claude_agent_sdk import (
tool,
create_sdk_mcp_server,
ClaudeSDKClient,
ClaudeAgentOptions,
AssistantMessage,
TextBlock,
)
@tool("get_weather", "Get the current weather for a location", {"location": str})
async def get_weather(args):
location = args["location"]
return {"content": [{"type": "text", "text": f"The weather in {location} is sunny and 72°F."}]}
server = create_sdk_mcp_server("weather-tools", tools=[get_weather])
async def main():
options = ClaudeAgentOptions(mcp_servers={"weather": server})
async with ClaudeSDKClient(options=options) as client:
await client.query("What's the weather in Paris?")
async for message in client.receive_response():
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
anyio.run(main)
```
---
## Hooks
### After Tool Use Hook
Log file changes after any edit:
```python
import anyio
from datetime import datetime
from claude_agent_sdk import query, ClaudeAgentOptions, HookMatcher, ResultMessage
async def log_file_change(input_data, tool_use_id, context):
file_path = input_data.get('tool_input', {}).get('file_path', 'unknown')
with open('./audit.log', 'a') as f:
f.write(f"{datetime.now()}: modified {file_path}\n")
return {}
async def main():
async for message in query(
prompt="Refactor utils.py to improve readability",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Edit", "Write"],
permission_mode="acceptEdits",
hooks={
"PostToolUse": [HookMatcher(matcher="Edit|Write", hooks=[log_file_change])]
}
)
):
if isinstance(message, ResultMessage):
print(message.result)
anyio.run(main)
```
---
## Subagents
```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition, ResultMessage
async def main():
async for message in query(
prompt="Use the code-reviewer agent to review this codebase",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep", "Agent"],
agents={
"code-reviewer": AgentDefinition(
description="Expert code reviewer for quality and security reviews.",
prompt="Analyze code quality and suggest improvements.",
tools=["Read", "Glob", "Grep"]
)
}
)
):
if isinstance(message, ResultMessage):
print(message.result)
anyio.run(main)
```
---
## MCP Server Integration
### Browser Automation (Playwright)
```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async def main():
async for message in query(
prompt="Open example.com and describe what you see",
options=ClaudeAgentOptions(
mcp_servers={
"playwright": {"command": "npx", "args": ["@playwright/mcp@latest"]}
}
)
):
if isinstance(message, ResultMessage):
print(message.result)
anyio.run(main)
```
### Database Access (PostgreSQL)
```python
import os
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async def main():
async for message in query(
prompt="Show me the top 10 users by order count",
options=ClaudeAgentOptions(
mcp_servers={
"postgres": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres"],
"env": {"DATABASE_URL": os.environ["DATABASE_URL"]}
}
}
)
):
if isinstance(message, ResultMessage):
print(message.result)
anyio.run(main)
```
---
## Permission Modes
```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions
async def main():
# Default: prompt for dangerous operations
async for message in query(
prompt="Delete all test files",
options=ClaudeAgentOptions(
allowed_tools=["Bash"],
permission_mode="default" # Will prompt before deleting
)
):
pass
# Plan: agent creates a plan before making changes
async for message in query(
prompt="Refactor the auth system",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Edit"],
permission_mode="plan"
)
):
pass
# Accept edits: auto-accept file edits
async for message in query(
prompt="Refactor this module",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Edit"],
permission_mode="acceptEdits"
)
):
pass
# Bypass: skip all prompts (use with caution)
async for message in query(
prompt="Set up the development environment",
options=ClaudeAgentOptions(
allowed_tools=["Bash", "Write"],
permission_mode="bypassPermissions",
allow_dangerously_skip_permissions=True
)
):
pass
anyio.run(main)
```
---
## Error Recovery
```python
import anyio
from claude_agent_sdk import (
query,
ClaudeAgentOptions,
CLINotFoundError,
CLIConnectionError,
ProcessError,
ResultMessage,
)
async def run_with_recovery():
try:
async for message in query(
prompt="Fix the failing tests",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Edit", "Bash"],
max_turns=10
)
):
if isinstance(message, ResultMessage):
print(message.result)
except CLINotFoundError:
print("Claude Code CLI not found. Install with: pip install claude-agent-sdk")
except CLIConnectionError as e:
print(f"Connection error: {e}")
except ProcessError as e:
print(f"Process error: {e}")
anyio.run(run_with_recovery)
```
---
## Session Resumption
```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage, SystemMessage
async def main():
session_id = None
# First query: capture the session ID
async for message in query(
prompt="Read the authentication module",
options=ClaudeAgentOptions(allowed_tools=["Read", "Glob"])
):
if isinstance(message, SystemMessage) and message.subtype == "init":
session_id = message.session_id
# Resume with full context from the first query
async for message in query(
prompt="Now find all places that call it", # "it" = auth module
options=ClaudeAgentOptions(resume=session_id)
):
if isinstance(message, ResultMessage):
print(message.result)
anyio.run(main)
```
---
## Custom System Prompt
```python
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async def main():
async for message in query(
prompt="Review this code",
options=ClaudeAgentOptions(
allowed_tools=["Read", "Glob", "Grep"],
system_prompt="""You are a senior code reviewer focused on:
1. Security vulnerabilities
2. Performance issues
3. Code maintainability
Always provide specific line numbers and suggestions for improvement."""
)
):
if isinstance(message, ResultMessage):
print(message.result)
anyio.run(main)
```

404
python/claude-api/README.md Normal file
View File

@@ -0,0 +1,404 @@
# Claude API — Python
## Installation
```bash
pip install anthropic
```
## Client Initialization
```python
import anthropic
# Default (uses ANTHROPIC_API_KEY env var)
client = anthropic.Anthropic()
# Explicit API key
client = anthropic.Anthropic(api_key="your-api-key")
# Async client
async_client = anthropic.AsyncAnthropic()
```
---
## Basic Message Request
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": "What is the capital of France?"}
]
)
print(response.content[0].text)
```
---
## System Prompts
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
system="You are a helpful coding assistant. Always provide examples in Python.",
messages=[{"role": "user", "content": "How do I read a JSON file?"}]
)
```
---
## Vision (Images)
### Base64
```python
import base64
with open("image.png", "rb") as f:
image_data = base64.standard_b64encode(f.read()).decode("utf-8")
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": image_data
}
},
{"type": "text", "text": "What's in this image?"}
]
}]
)
```
### URL
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "url",
"url": "https://example.com/image.png"
}
},
{"type": "text", "text": "Describe this image"}
]
}]
)
```
---
## Prompt Caching
Cache large context to reduce costs (up to 90% savings).
### Automatic Caching (Recommended)
Use top-level `cache_control` to automatically cache the last cacheable block in the request — no need to annotate individual content blocks:
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
cache_control={"type": "ephemeral"}, # auto-caches the last cacheable block
system="You are an expert on this large document...",
messages=[{"role": "user", "content": "Summarize the key points"}]
)
```
### Manual Cache Control
For fine-grained control, add `cache_control` to specific content blocks:
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
system=[{
"type": "text",
"text": "You are an expert on this large document...",
"cache_control": {"type": "ephemeral"} # default TTL is 5 minutes
}],
messages=[{"role": "user", "content": "Summarize the key points"}]
)
# With explicit TTL (time-to-live)
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
system=[{
"type": "text",
"text": "You are an expert on this large document...",
"cache_control": {"type": "ephemeral", "ttl": "1h"} # 1 hour TTL
}],
messages=[{"role": "user", "content": "Summarize the key points"}]
)
```
---
## Extended Thinking
> **Opus 4.6 and Sonnet 4.6:** Use adaptive thinking. `budget_tokens` is deprecated on both Opus 4.6 and Sonnet 4.6.
> **Older models:** Use `thinking: {type: "enabled", budget_tokens: N}` (must be < `max_tokens`, min 1024).
```python
# Opus 4.6: adaptive thinking (recommended)
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=16000,
thinking={"type": "adaptive"},
output_config={"effort": "high"}, # low | medium | high | max
messages=[{"role": "user", "content": "Solve this step by step..."}]
)
# Access thinking and response
for block in response.content:
if block.type == "thinking":
print(f"Thinking: {block.thinking}")
elif block.type == "text":
print(f"Response: {block.text}")
```
---
## Error Handling
```python
import anthropic
try:
response = client.messages.create(...)
except anthropic.BadRequestError as e:
print(f"Bad request: {e.message}")
except anthropic.AuthenticationError:
print("Invalid API key")
except anthropic.PermissionDeniedError:
print("API key lacks required permissions")
except anthropic.NotFoundError:
print("Invalid model or endpoint")
except anthropic.RateLimitError as e:
retry_after = int(e.response.headers.get("retry-after", "60"))
print(f"Rate limited. Retry after {retry_after}s.")
except anthropic.APIStatusError as e:
if e.status_code >= 500:
print(f"Server error ({e.status_code}). Retry later.")
else:
print(f"API error: {e.message}")
except anthropic.APIConnectionError:
print("Network error. Check internet connection.")
```
---
## Multi-Turn Conversations
The API is stateless — send the full conversation history each time.
```python
class ConversationManager:
"""Manage multi-turn conversations with the Claude API."""
def __init__(self, client: anthropic.Anthropic, model: str, system: str = None):
self.client = client
self.model = model
self.system = system
self.messages = []
def send(self, user_message: str, **kwargs) -> str:
"""Send a message and get a response."""
self.messages.append({"role": "user", "content": user_message})
response = self.client.messages.create(
model=self.model,
max_tokens=kwargs.get("max_tokens", 1024),
system=self.system,
messages=self.messages,
**kwargs
)
assistant_message = response.content[0].text
self.messages.append({"role": "assistant", "content": assistant_message})
return assistant_message
# Usage
conversation = ConversationManager(
client=anthropic.Anthropic(),
model="claude-opus-4-6",
system="You are a helpful assistant."
)
response1 = conversation.send("My name is Alice.")
response2 = conversation.send("What's my name?") # Claude remembers "Alice"
```
**Rules:**
- Messages must alternate between `user` and `assistant`
- First message must be `user`
---
### Compaction (long conversations)
> **Beta, Opus 4.6 only.** When conversations approach the 200K context window, compaction automatically summarizes earlier context server-side. The API returns a `compaction` block; you must pass it back on subsequent requests — append `response.content`, not just the text.
```python
import anthropic
client = anthropic.Anthropic()
messages = []
def chat(user_message: str) -> str:
messages.append({"role": "user", "content": user_message})
response = client.beta.messages.create(
betas=["compact-2026-01-12"],
model="claude-opus-4-6",
max_tokens=4096,
messages=messages,
context_management={
"edits": [{"type": "compact_20260112"}]
}
)
# Append full content — compaction blocks must be preserved
messages.append({"role": "assistant", "content": response.content})
return next(block.text for block in response.content if block.type == "text")
# Compaction triggers automatically when context grows large
print(chat("Help me build a Python web scraper"))
print(chat("Add support for JavaScript-rendered pages"))
print(chat("Now add rate limiting and error handling"))
```
---
## Stop Reasons
The `stop_reason` field in the response indicates why the model stopped generating:
| Value | Meaning |
|-------|---------|
| `end_turn` | Claude finished its response naturally |
| `max_tokens` | Hit the `max_tokens` limit — increase it or use streaming |
| `stop_sequence` | Hit a custom stop sequence |
| `tool_use` | Claude wants to call a tool — execute it and continue |
| `pause_turn` | Model paused and can be resumed (agentic flows) |
| `refusal` | Claude refused for safety reasons — output may not match your schema |
---
## Cost Optimization Strategies
### 1. Use Prompt Caching for Repeated Context
```python
# Automatic caching (simplest — caches the last cacheable block)
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
cache_control={"type": "ephemeral"},
system=large_document_text, # e.g., 50KB of context
messages=[{"role": "user", "content": "Summarize the key points"}]
)
# First request: full cost
# Subsequent requests: ~90% cheaper for cached portion
```
### 2. Choose the Right Model
```python
# Default to Opus for most tasks
response = client.messages.create(
model="claude-opus-4-6", # $5.00/$25.00 per 1M tokens
max_tokens=1024,
messages=[{"role": "user", "content": "Explain quantum computing"}]
)
# Use Sonnet for high-volume production workloads
standard_response = client.messages.create(
model="claude-sonnet-4-6", # $3.00/$15.00 per 1M tokens
max_tokens=1024,
messages=[{"role": "user", "content": "Summarize this document"}]
)
# Use Haiku only for simple, speed-critical tasks
simple_response = client.messages.create(
model="claude-haiku-4-5", # $1.00/$5.00 per 1M tokens
max_tokens=256,
messages=[{"role": "user", "content": "Classify this as positive or negative"}]
)
```
### 3. Use Token Counting Before Requests
```python
count_response = client.messages.count_tokens(
model="claude-opus-4-6",
messages=messages,
system=system
)
estimated_input_cost = count_response.input_tokens * 0.000005 # $5/1M tokens
print(f"Estimated input cost: ${estimated_input_cost:.4f}")
```
---
## Retry with Exponential Backoff
> **Note:** The Anthropic SDK automatically retries rate limit (429) and server errors (5xx) with exponential backoff. You can configure this with `max_retries` (default: 2). Only implement custom retry logic if you need behavior beyond what the SDK provides.
```python
import time
import random
import anthropic
def call_with_retry(
client: anthropic.Anthropic,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
**kwargs
):
"""Call the API with exponential backoff retry."""
last_exception = None
for attempt in range(max_retries):
try:
return client.messages.create(**kwargs)
except anthropic.RateLimitError as e:
last_exception = e
except anthropic.APIStatusError as e:
if e.status_code >= 500:
last_exception = e
else:
raise # Client errors (4xx except 429) should not be retried
delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
print(f"Retry {attempt + 1}/{max_retries} after {delay:.1f}s")
time.sleep(delay)
raise last_exception
```

182
python/claude-api/batches.md Normal file
View File

@@ -0,0 +1,182 @@
# Message Batches API — Python
The Batches API (`POST /v1/messages/batches`) processes Messages API requests asynchronously at 50% of standard prices.
## Key Facts
- Up to 100,000 requests or 256 MB per batch
- Most batches complete within 1 hour; maximum 24 hours
- Results available for 29 days after creation
- 50% cost reduction on all token usage
- All Messages API features supported (vision, tools, caching, etc.)
---
## Create a Batch
```python
import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
client = anthropic.Anthropic()
message_batch = client.messages.batches.create(
requests=[
Request(
custom_id="request-1",
params=MessageCreateParamsNonStreaming(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Summarize climate change impacts"}]
)
),
Request(
custom_id="request-2",
params=MessageCreateParamsNonStreaming(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Explain quantum computing basics"}]
)
),
]
)
print(f"Batch ID: {message_batch.id}")
print(f"Status: {message_batch.processing_status}")
```
---
## Poll for Completion
```python
import time
while True:
batch = client.messages.batches.retrieve(message_batch.id)
if batch.processing_status == "ended":
break
print(f"Status: {batch.processing_status}, processing: {batch.request_counts.processing}")
time.sleep(60)
print("Batch complete!")
print(f"Succeeded: {batch.request_counts.succeeded}")
print(f"Errored: {batch.request_counts.errored}")
```
---
## Retrieve Results
> **Note:** Examples below use `match/case` syntax, requiring Python 3.10+. For earlier versions, use `if/elif` chains instead.
```python
for result in client.messages.batches.results(message_batch.id):
match result.result.type:
case "succeeded":
print(f"[{result.custom_id}] {result.result.message.content[0].text[:100]}")
case "errored":
if result.result.error.type == "invalid_request":
print(f"[{result.custom_id}] Validation error - fix request and retry")
else:
print(f"[{result.custom_id}] Server error - safe to retry")
case "canceled":
print(f"[{result.custom_id}] Canceled")
case "expired":
print(f"[{result.custom_id}] Expired - resubmit")
```
---
## Cancel a Batch
```python
cancelled = client.messages.batches.cancel(message_batch.id)
print(f"Status: {cancelled.processing_status}") # "canceling"
```
---
## Batch with Prompt Caching
```python
shared_system = [
{"type": "text", "text": "You are a literary analyst."},
{
"type": "text",
"text": large_document_text, # Shared across all requests
"cache_control": {"type": "ephemeral"}
}
]
message_batch = client.messages.batches.create(
requests=[
Request(
custom_id=f"analysis-{i}",
params=MessageCreateParamsNonStreaming(
model="claude-opus-4-6",
max_tokens=1024,
system=shared_system,
messages=[{"role": "user", "content": question}]
)
)
for i, question in enumerate(questions)
]
)
```
---
## Full End-to-End Example
```python
import anthropic
import time
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
client = anthropic.Anthropic()
# 1. Prepare requests
items_to_classify = [
"The product quality is excellent!",
"Terrible customer service, never again.",
"It's okay, nothing special.",
]
requests = [
Request(
custom_id=f"classify-{i}",
params=MessageCreateParamsNonStreaming(
model="claude-haiku-4-5",
max_tokens=50,
messages=[{
"role": "user",
"content": f"Classify as positive/negative/neutral (one word): {text}"
}]
)
)
for i, text in enumerate(items_to_classify)
]
# 2. Create batch
batch = client.messages.batches.create(requests=requests)
print(f"Created batch: {batch.id}")
# 3. Wait for completion
while True:
batch = client.messages.batches.retrieve(batch.id)
if batch.processing_status == "ended":
break
time.sleep(10)
# 4. Collect results
results = {}
for result in client.messages.batches.results(batch.id):
if result.result.type == "succeeded":
results[result.custom_id] = result.result.message.content[0].text
for custom_id, classification in sorted(results.items()):
print(f"{custom_id}: {classification}")
```

162
python/claude-api/files-api.md Normal file
View File

@@ -0,0 +1,162 @@
# Files API — Python
The Files API uploads files for use in Messages API requests. Reference files via `file_id` in content blocks, avoiding re-uploads across multiple API calls.
**Beta:** Pass `betas=["files-api-2025-04-14"]` in your API calls (the SDK sets the required header automatically).
## Key Facts
- Maximum file size: 500 MB
- Total storage: 100 GB per organization
- Files persist until deleted
- File operations (upload, list, delete) are free; content used in messages is billed as input tokens
- Not available on Amazon Bedrock or Google Vertex AI
---
## Upload a File
```python
import anthropic
client = anthropic.Anthropic()
uploaded = client.beta.files.upload(
file=("report.pdf", open("report.pdf", "rb"), "application/pdf"),
)
print(f"File ID: {uploaded.id}")
print(f"Size: {uploaded.size_bytes} bytes")
```
---
## Use a File in Messages
### PDF / Text Document
```python
response = client.beta.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": "Summarize the key findings in this report."},
{
"type": "document",
"source": {"type": "file", "file_id": uploaded.id},
"title": "Q4 Report", # optional
"citations": {"enabled": True} # optional, enables citations
}
]
}],
betas=["files-api-2025-04-14"],
)
print(response.content[0].text)
```
### Image
```python
image_file = client.beta.files.upload(
file=("photo.png", open("photo.png", "rb"), "image/png"),
)
response = client.beta.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": "What's in this image?"},
{
"type": "image",
"source": {"type": "file", "file_id": image_file.id}
}
]
}],
betas=["files-api-2025-04-14"],
)
```
---
## Manage Files
### List Files
```python
files = client.beta.files.list()
for f in files.data:
print(f"{f.id}: {f.filename} ({f.size_bytes} bytes)")
```
### Get File Metadata
```python
file_info = client.beta.files.retrieve_metadata("file_011CNha8iCJcU1wXNR6q4V8w")
print(f"Filename: {file_info.filename}")
print(f"MIME type: {file_info.mime_type}")
```
### Delete a File
```python
client.beta.files.delete("file_011CNha8iCJcU1wXNR6q4V8w")
```
### Download a File
Only files created by the code execution tool or skills can be downloaded (not user-uploaded files).
```python
file_content = client.beta.files.download("file_011CNha8iCJcU1wXNR6q4V8w")
file_content.write_to_file("output.txt")
```
---
## Full End-to-End Example
Upload a document once, ask multiple questions about it:
```python
import anthropic
client = anthropic.Anthropic()
# 1. Upload once
uploaded = client.beta.files.upload(
file=("contract.pdf", open("contract.pdf", "rb"), "application/pdf"),
)
print(f"Uploaded: {uploaded.id}")
# 2. Ask multiple questions using the same file_id
questions = [
"What are the key terms and conditions?",
"What is the termination clause?",
"Summarize the payment schedule.",
]
for question in questions:
response = client.beta.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": [
{"type": "text", "text": question},
{
"type": "document",
"source": {"type": "file", "file_id": uploaded.id}
}
]
}],
betas=["files-api-2025-04-14"],
)
print(f"\nQ: {question}")
print(f"A: {response.content[0].text[:200]}")
# 3. Clean up when done
client.beta.files.delete(uploaded.id)
```

162
python/claude-api/streaming.md Normal file
View File

@@ -0,0 +1,162 @@
# Streaming — Python
## Quick Start
```python
with client.messages.stream(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Write a story"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
```
### Async
```python
async with async_client.messages.stream(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Write a story"}]
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)
```
---
## Handling Different Content Types
Claude may return text, thinking blocks, or tool use. Handle each appropriately:
> **Opus 4.6:** Use `thinking: {type: "adaptive"}`. On older models, use `thinking: {type: "enabled", budget_tokens: N}` instead.
```python
with client.messages.stream(
model="claude-opus-4-6",
max_tokens=16000,
thinking={"type": "adaptive"},
messages=[{"role": "user", "content": "Analyze this problem"}]
) as stream:
for event in stream:
if event.type == "content_block_start":
if event.content_block.type == "thinking":
print("\n[Thinking...]")
elif event.content_block.type == "text":
print("\n[Response:]")
elif event.type == "content_block_delta":
if event.delta.type == "thinking_delta":
print(event.delta.thinking, end="", flush=True)
elif event.delta.type == "text_delta":
print(event.delta.text, end="", flush=True)
```
---
## Streaming with Tool Use
The Python tool runner currently returns complete messages. Use streaming for individual API calls within a manual loop if you need per-token streaming with tools:
```python
with client.messages.stream(
model="claude-opus-4-6",
max_tokens=4096,
tools=tools,
messages=messages
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
response = stream.get_final_message()
# Continue with tool execution if response.stop_reason == "tool_use"
```
---
## Getting the Final Message
```python
with client.messages.stream(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Hello"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
# Get full message after streaming
final_message = stream.get_final_message()
print(f"\n\nTokens used: {final_message.usage.output_tokens}")
```
---
## Streaming with Progress Updates
```python
def stream_with_progress(client, **kwargs):
"""Stream a response with progress updates."""
total_tokens = 0
content_parts = []
with client.messages.stream(**kwargs) as stream:
for event in stream:
if event.type == "content_block_delta":
if event.delta.type == "text_delta":
text = event.delta.text
content_parts.append(text)
print(text, end="", flush=True)
elif event.type == "message_delta":
if event.usage and event.usage.output_tokens is not None:
total_tokens = event.usage.output_tokens
final_message = stream.get_final_message()
print(f"\n\n[Tokens used: {total_tokens}]")
return "".join(content_parts)
```
---
## Error Handling in Streams
```python
try:
with client.messages.stream(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Write a story"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
except anthropic.APIConnectionError:
print("\nConnection lost. Please retry.")
except anthropic.RateLimitError:
print("\nRate limited. Please wait and retry.")
except anthropic.APIStatusError as e:
print(f"\nAPI error: {e.status_code}")
```
---
## Stream Event Types
| Event Type | Description | When it fires |
| --------------------- | --------------------------- | --------------------------------- |
| `message_start` | Contains message metadata | Once at the beginning |
| `content_block_start` | New content block beginning | When a text/tool_use block starts |
| `content_block_delta` | Incremental content update | For each token/chunk |
| `content_block_stop` | Content block complete | When a block finishes |
| `message_delta` | Message-level updates | Contains `stop_reason`, usage |
| `message_stop` | Message complete | Once at the end |
## Best Practices
1. **Always flush output** — Use `flush=True` to show tokens immediately
2. **Handle partial responses** — If the stream is interrupted, you may have incomplete content
3. **Track token usage** — The `message_delta` event contains usage information
4. **Use timeouts** — Set appropriate timeouts for your application
5. **Default to streaming** — Use `.get_final_message()` to get the complete response even when streaming, giving you timeout protection without needing to handle individual events

587
python/claude-api/tool-use.md Normal file
View File

@@ -0,0 +1,587 @@
# Tool Use — Python
For conceptual overview (tool definitions, tool choice, tips), see [shared/tool-use-concepts.md](../../shared/tool-use-concepts.md).
## Tool Runner (Recommended)
**Beta:** The tool runner is in beta in the Python SDK.
Use the `@beta_tool` decorator to define tools as typed functions, then pass them to `client.beta.messages.tool_runner()`:
```python
import anthropic
from anthropic import beta_tool
client = anthropic.Anthropic()
@beta_tool
def get_weather(location: str, unit: str = "celsius") -> str:
"""Get current weather for a location.
Args:
location: City and state, e.g., San Francisco, CA.
unit: Temperature unit, either "celsius" or "fahrenheit".
"""
# Your implementation here
return f"72°F and sunny in {location}"
# The tool runner handles the agentic loop automatically
runner = client.beta.messages.tool_runner(
model="claude-opus-4-6",
max_tokens=4096,
tools=[get_weather],
messages=[{"role": "user", "content": "What's the weather in Paris?"}],
)
# Each iteration yields a BetaMessage; iteration stops when Claude is done
for message in runner:
print(message)
```
For async usage, use `@beta_async_tool` with `async def` functions.
**Key benefits of the tool runner:**
- No manual loop — the SDK handles calling tools and feeding results back
- Type-safe tool inputs via decorators
- Tool schemas are generated automatically from function signatures
- Iteration stops automatically when Claude has no more tool calls
---
## MCP Tool Conversion Helpers
**Beta.** Convert [MCP (Model Context Protocol)](https://modelcontextprotocol.io/) tools, prompts, and resources to Anthropic API types for use with the tool runner. Requires `pip install anthropic[mcp]` (Python 3.10+).
> **Note:** The Claude API also supports an `mcp_servers` parameter that lets Claude connect directly to remote MCP servers. Use these helpers instead when you need local MCP servers, prompts, resources, or more control over the MCP connection.
### MCP Tools with Tool Runner
```python
from anthropic import AsyncAnthropic
from anthropic.lib.tools.mcp import async_mcp_tool
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
client = AsyncAnthropic()
async with stdio_client(StdioServerParameters(command="mcp-server")) as (read, write):
async with ClientSession(read, write) as mcp_client:
await mcp_client.initialize()
tools_result = await mcp_client.list_tools()
runner = await client.beta.messages.tool_runner(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Use the available tools"}],
tools=[async_mcp_tool(t, mcp_client) for t in tools_result.tools],
)
async for message in runner:
print(message)
```
For sync usage, use `mcp_tool` instead of `async_mcp_tool`.
### MCP Prompts
```python
from anthropic.lib.tools.mcp import mcp_message
prompt = await mcp_client.get_prompt(name="my-prompt")
response = await client.beta.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[mcp_message(m) for m in prompt.messages],
)
```
### MCP Resources as Content
```python
from anthropic.lib.tools.mcp import mcp_resource_to_content
resource = await mcp_client.read_resource(uri="file:///path/to/doc.txt")
response = await client.beta.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": [
mcp_resource_to_content(resource),
{"type": "text", "text": "Summarize this document"},
],
}],
)
```
### Upload MCP Resources as Files
```python
from anthropic.lib.tools.mcp import mcp_resource_to_file
resource = await mcp_client.read_resource(uri="file:///path/to/data.json")
uploaded = await client.beta.files.upload(file=mcp_resource_to_file(resource))
```
Conversion functions raise `UnsupportedMCPValueError` if an MCP value cannot be converted (e.g., unsupported content types like audio, unsupported MIME types).
---
## Manual Agentic Loop
Use this when you need fine-grained control over the loop (e.g., custom logging, conditional tool execution, human-in-the-loop approval):
```python
import anthropic
client = anthropic.Anthropic()
tools = [...] # Your tool definitions
messages = [{"role": "user", "content": user_input}]
# Agentic loop: keep going until Claude stops calling tools
while True:
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=4096,
tools=tools,
messages=messages
)
# If Claude is done (no more tool calls), break
if response.stop_reason == "end_turn":
break
# Server-side tool hit iteration limit; re-send to continue
if response.stop_reason == "pause_turn":
messages = [
{"role": "user", "content": user_input},
{"role": "assistant", "content": response.content},
]
continue
# Extract tool use blocks from the response
tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
# Append assistant's response (including tool_use blocks)
messages.append({"role": "assistant", "content": response.content})
# Execute each tool and collect results
tool_results = []
for tool in tool_use_blocks:
result = execute_tool(tool.name, tool.input) # Your implementation
tool_results.append({
"type": "tool_result",
"tool_use_id": tool.id, # Must match the tool_use block's id
"content": result
})
# Append tool results as a user message
messages.append({"role": "user", "content": tool_results})
# Final response text
final_text = next(b.text for b in response.content if b.type == "text")
```
---
## Handling Tool Results
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
tools=tools,
messages=[{"role": "user", "content": "What's the weather in Paris?"}]
)
for block in response.content:
if block.type == "tool_use":
tool_name = block.name
tool_input = block.input
tool_use_id = block.id
result = execute_tool(tool_name, tool_input)
followup = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
tools=tools,
messages=[
{"role": "user", "content": "What's the weather in Paris?"},
{"role": "assistant", "content": response.content},
{
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": result
}]
}
]
)
```
---
## Multiple Tool Calls
```python
tool_results = []
for block in response.content:
if block.type == "tool_use":
result = execute_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
# Send all results back at once
if tool_results:
followup = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
tools=tools,
messages=[
*previous_messages,
{"role": "assistant", "content": response.content},
{"role": "user", "content": tool_results}
]
)
```
---
## Error Handling in Tool Results
```python
tool_result = {
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": "Error: Location 'xyz' not found. Please provide a valid city name.",
"is_error": True
}
```
---
## Tool Choice
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
tools=tools,
tool_choice={"type": "tool", "name": "get_weather"}, # Force specific tool
messages=[{"role": "user", "content": "What's the weather in Paris?"}]
)
```
---
## Code Execution
### Basic Usage
```python
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=4096,
messages=[{
"role": "user",
"content": "Calculate the mean and standard deviation of [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]"
}],
tools=[{
"type": "code_execution_20260120",
"name": "code_execution"
}]
)
for block in response.content:
if block.type == "text":
print(block.text)
elif block.type == "bash_code_execution_tool_result":
print(f"stdout: {block.content.stdout}")
```
### Upload Files for Analysis
```python
# 1. Upload a file
uploaded = client.beta.files.upload(file=open("sales_data.csv", "rb"))
# 2. Pass to code execution via container_upload block
# Code execution is GA; Files API is still beta (pass via extra_headers)
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=4096,
extra_headers={"anthropic-beta": "files-api-2025-04-14"},
messages=[{
"role": "user",
"content": [
{"type": "text", "text": "Analyze this sales data. Show trends and create a visualization."},
{"type": "container_upload", "file_id": uploaded.id}
]
}],
tools=[{"type": "code_execution_20260120", "name": "code_execution"}]
)
```
### Retrieve Generated Files
```python
import os
OUTPUT_DIR = "./claude_outputs"
os.makedirs(OUTPUT_DIR, exist_ok=True)
for block in response.content:
if block.type == "bash_code_execution_tool_result":
result = block.content
if result.type == "bash_code_execution_result" and result.content:
for file_ref in result.content:
if file_ref.type == "bash_code_execution_output":
metadata = client.beta.files.retrieve_metadata(file_ref.file_id)
file_content = client.beta.files.download(file_ref.file_id)
# Use basename to prevent path traversal; validate result
safe_name = os.path.basename(metadata.filename)
if not safe_name or safe_name in (".", ".."):
print(f"Skipping invalid filename: {metadata.filename}")
continue
output_path = os.path.join(OUTPUT_DIR, safe_name)
file_content.write_to_file(output_path)
print(f"Saved: {output_path}")
```
### Container Reuse
```python
# First request: set up environment
response1 = client.messages.create(
model="claude-opus-4-6",
max_tokens=4096,
messages=[{"role": "user", "content": "Install tabulate and create data.json with sample data"}],
tools=[{"type": "code_execution_20260120", "name": "code_execution"}]
)
# Get container ID from response
container_id = response1.container.id
# Second request: reuse the same container
response2 = client.messages.create(
container=container_id,
model="claude-opus-4-6",
max_tokens=4096,
messages=[{"role": "user", "content": "Read data.json and display as a formatted table"}],
tools=[{"type": "code_execution_20260120", "name": "code_execution"}]
)
```
### Response Structure
```python
for block in response.content:
if block.type == "text":
print(block.text) # Claude's explanation
elif block.type == "server_tool_use":
print(f"Running: {block.name} - {block.input}") # What Claude is doing
elif block.type == "bash_code_execution_tool_result":
result = block.content
if result.type == "bash_code_execution_result":
if result.return_code == 0:
print(f"Output: {result.stdout}")
else:
print(f"Error: {result.stderr}")
else:
print(f"Tool error: {result.error_code}")
elif block.type == "text_editor_code_execution_tool_result":
print(f"File operation: {block.content}")
```
---
## Memory Tool
### Basic Usage
```python
import anthropic
client = anthropic.Anthropic()
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=2048,
messages=[{"role": "user", "content": "Remember that my preferred language is Python."}],
tools=[{"type": "memory_20250818", "name": "memory"}],
)
```
### SDK Memory Helper
Subclass `BetaAbstractMemoryTool`:
```python
from anthropic.lib.tools import BetaAbstractMemoryTool
class MyMemoryTool(BetaAbstractMemoryTool):
def view(self, command): ...
def create(self, command): ...
def str_replace(self, command): ...
def insert(self, command): ...
def delete(self, command): ...
def rename(self, command): ...
memory = MyMemoryTool()
# Use with tool runner
runner = client.beta.messages.tool_runner(
model="claude-opus-4-6",
max_tokens=2048,
tools=[memory],
messages=[{"role": "user", "content": "Remember my preferences"}],
)
for message in runner:
print(message)
```
For full implementation examples, use WebFetch:
- `https://github.com/anthropics/anthropic-sdk-python/blob/main/examples/memory/basic.py`
---
## Structured Outputs
### JSON Outputs (Pydantic — Recommended)
```python
from pydantic import BaseModel
from typing import List
import anthropic
class ContactInfo(BaseModel):
name: str
email: str
plan: str
interests: List[str]
demo_requested: bool
client = anthropic.Anthropic()
response = client.messages.parse(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": "Extract: Jane Doe (jane@co.com) wants Enterprise, interested in API and SDKs, wants a demo."
}],
output_format=ContactInfo,
)
# response.parsed_output is a validated ContactInfo instance
contact = response.parsed_output
print(contact.name) # "Jane Doe"
print(contact.interests) # ["API", "SDKs"]
```
### Raw Schema
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{
"role": "user",
"content": "Extract info: John Smith (john@example.com) wants the Enterprise plan."
}],
output_config={
"format": {
"type": "json_schema",
"schema": {
"type": "object",
"properties": {
"name": {"type": "string"},
"email": {"type": "string"},
"plan": {"type": "string"},
"demo_requested": {"type": "boolean"}
},
"required": ["name", "email", "plan", "demo_requested"],
"additionalProperties": False
}
}
}
)
import json
data = json.loads(response.content[0].text)
```
### Strict Tool Use
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Book a flight to Tokyo for 2 passengers on March 15"}],
tools=[{
"name": "book_flight",
"description": "Book a flight to a destination",
"strict": True,
"input_schema": {
"type": "object",
"properties": {
"destination": {"type": "string"},
"date": {"type": "string", "format": "date"},
"passengers": {"type": "integer", "enum": [1, 2, 3, 4, 5, 6, 7, 8]}
},
"required": ["destination", "date", "passengers"],
"additionalProperties": False
}
}]
)
```
### Using Both Together
```python
response = client.messages.create(
model="claude-opus-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Plan a trip to Paris next month"}],
output_config={
"format": {
"type": "json_schema",
"schema": {
"type": "object",
"properties": {
"summary": {"type": "string"},
"next_steps": {"type": "array", "items": {"type": "string"}}
},
"required": ["summary", "next_steps"],
"additionalProperties": False
}
}
},
tools=[{
"name": "search_flights",
"description": "Search for available flights",
"strict": True,
"input_schema": {
"type": "object",
"properties": {
"destination": {"type": "string"},
"date": {"type": "string", "format": "date"}
},
"required": ["destination", "date"],
"additionalProperties": False
}
}]
)
```