mirrors/claude-code-sdk-python
1
0
Fork 0
mirror of https://github.com/anthropics/claude-code-sdk-python.git synced 2025-12-23 09:19:52 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Update changelog, readme, and examples for custom tools and hooks (#162)

Browse source
This commit is contained in:
Dickson Tsai 2025-09-12 08:14:34 -07:00 • committed by GitHub
parent 0aab45be7d
commit db6f167280
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 152 additions and 155 deletions
Download patch file Download diff file Expand all files Collapse all files

7
CHANGELOG.md
View file

@ -1,5 +1,12 @@
# Changelog
## 0.0.22
- Introduce custom tools, implemented as in-process MCP servers.
- Introduce hooks.
- Update internal `Transport` class to lower-level interface.
- `ClaudeSDKClient` can no longer be run in different async contexts.
## 0.0.19
- Add `ClaudeCodeOptions.add_dirs` for `--add-dir`

93
README.md
View file

@ -1,6 +1,6 @@
# Claude Code SDK for Python
Python SDK for Claude Code. See the [Claude Code SDK documentation](https://docs.anthropic.com/en/docs/claude-code/sdk) for more information.
Python SDK for Claude Code. See the [Claude Code SDK documentation](https://docs.anthropic.com/en/docs/claude-code/sdk/sdk-python) for more information.
## Installation
@ -26,9 +26,9 @@ async def main():
anyio.run(main)
```
## Usage
## Basic Usage: query()
### Basic Query
`query()` is an async function for querying Claude Code. It returns an `AsyncIterator` of response messages. See [src/claude_code_sdk/query.py](src/claude_code_sdk/query.py).
```python
from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, TextBlock
@ -76,14 +76,25 @@ options = ClaudeCodeOptions(
)
```
### SDK MCP Servers (In-Process)
## ClaudeSDKClient
The SDK now supports in-process MCP servers that run directly within your Python application, eliminating the need for separate processes.
`ClaudeSDKClient` supports bidirectional, interactive conversations with Claude
Code. See [src/claude_code_sdk/client.py](src/claude_code_sdk/client.py).
Unlike `query()`, `ClaudeSDKClient` additionally enables **custom tools** and **hooks**, both of which can be defined as Python functions.
### Custom Tools (as In-Process SDK MCP Servers)
A **custom tool** is a Python function that you can offer to Claude, for Claude to invoke as needed.
Custom tools are implemented in-process MCP servers that run directly within your Python application, eliminating the need for separate processes that regular MCP servers require.
For an end-to-end example, see [MCP Calculator](examples/mcp_calculator.py).
#### Creating a Simple Tool
```python
from claude_code_sdk import tool, create_sdk_mcp_server
from claude_code_sdk import tool, create_sdk_mcp_server, ClaudeCodeOptions, ClaudeSDKClient
# Define a tool using the @tool decorator
@tool("greet", "Greet a user", {"name": str})
@ -103,11 +114,16 @@ server = create_sdk_mcp_server(
# Use it with Claude
options = ClaudeCodeOptions(
mcp_servers={"tools": server}
mcp_servers={"tools": server},
allowed_tools=["mcp__tools__greet"]
)
async for message in query(prompt="Greet Alice", options=options):
print(message)
async with ClaudeSDKClient(options=options) as client:
await client.query("Greet Alice")
# Extract and print response
async for msg in client.receive_response():
print(msg)
```
#### Benefits Over External MCP Servers
@ -161,19 +177,60 @@ options = ClaudeCodeOptions(
)
```
## API Reference
### Hooks
### `query(prompt, options=None)`
A **hook** is a Python function that the Claude Code *application* (*not* Claude) invokes at specific points of the Claude agent loop. Hooks can provide deterministic processing and automated feedback for Claude. Read more in [Claude Code Hooks Reference](https://docs.anthropic.com/en/docs/claude-code/hooks).
Main async function for querying Claude.
For more examples, see examples/hooks.py.
**Parameters:**
- `prompt` (str): The prompt to send to Claude
- `options` (ClaudeCodeOptions): Optional configuration
#### Example
**Returns:** AsyncIterator[Message] - Stream of response messages
```python
from claude_code_sdk import ClaudeCodeOptions, ClaudeSDKClient, HookMatcher
### Types
async def check_bash_command(input_data, tool_use_id, context):
tool_name = input_data["tool_name"]
tool_input = input_data["tool_input"]
if tool_name != "Bash":
return {}
command = tool_input.get("command", "")
block_patterns = ["foo.sh"]
for pattern in block_patterns:
if pattern in command:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": f"Command contains invalid pattern: {pattern}",
}
}
return {}
options = ClaudeCodeOptions(
allowed_tools=["Bash"],
hooks={
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[check_bash_command]),
],
}
)
async with ClaudeSDKClient(options=options) as client:
# Test 1: Command with forbidden pattern (will be blocked)
await client.query("Run the bash command: ./foo.sh --help")
async for msg in client.receive_response():
print(msg)
print("\n" + "=" * 50 + "\n")
# Test 2: Safe command that should work
await client.query("Run the bash command: echo 'Hello from hooks example!'")
async for msg in client.receive_response():
print(msg)
```
## Types
See [src/claude_code_sdk/types.py](src/claude_code_sdk/types.py) for complete type definitions:
- `ClaudeCodeOptions` - Configuration options
@ -212,6 +269,8 @@ See the [Claude Code documentation](https://docs.anthropic.com/en/docs/claude-co
See [examples/quick_start.py](examples/quick_start.py) for a complete working example.
See [examples/streaming_mode.py](examples/streaming_mode.py) for comprehensive examples involving `ClaudeSDKClient`. You can even run interactive examples in IPython from [examples/streaming_mode_ipython.py](examples/streaming_mode_ipython.py).
## License
MIT

67
e2e-tests/test_sdk_mcp_tools.py
View file

@ -9,14 +9,11 @@ from typing import Any
import pytest
from claude_code_sdk import (
AssistantMessage,
ClaudeCodeOptions,
ClaudeSDKClient,
UserMessage,
create_sdk_mcp_server,
tool,
)
from claude_code_sdk.types import ToolResultBlock, ToolUseBlock
@pytest.mark.e2e
@ -24,30 +21,30 @@ from claude_code_sdk.types import ToolResultBlock, ToolUseBlock
async def test_sdk_mcp_tool_execution():
"""Test that SDK MCP tools can be called and executed with allowed_tools."""
executions = []
@tool("echo", "Echo back the input text", {"text": str})
async def echo_tool(args: dict[str, Any]) -> dict[str, Any]:
"""Echo back whatever text is provided."""
executions.append("echo")
return {"content": [{"type": "text", "text": f"Echo: {args['text']}"}]}
server = create_sdk_mcp_server(
name="test",
version="1.0.0",
tools=[echo_tool],
)
options = ClaudeCodeOptions(
mcp_servers={"test": server},
allowed_tools=["mcp__test__echo"],
)
async with ClaudeSDKClient(options=options) as client:
await client.query("Call the mcp__test__echo tool with any text")
async for message in client.receive_response():
pass # Just consume messages
# Check if the actual Python function was called
assert "echo" in executions, "Echo tool function was not executed"
@ -57,37 +54,39 @@ async def test_sdk_mcp_tool_execution():
async def test_sdk_mcp_permission_enforcement():
"""Test that disallowed_tools prevents SDK MCP tool execution."""
executions = []
@tool("echo", "Echo back the input text", {"text": str})
async def echo_tool(args: dict[str, Any]) -> dict[str, Any]:
"""Echo back whatever text is provided."""
executions.append("echo")
return {"content": [{"type": "text", "text": f"Echo: {args['text']}"}]}
@tool("greet", "Greet a person by name", {"name": str})
async def greet_tool(args: dict[str, Any]) -> dict[str, Any]:
"""Greet someone by name."""
executions.append("greet")
return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]}
server = create_sdk_mcp_server(
name="test",
version="1.0.0",
tools=[echo_tool, greet_tool],
)
options = ClaudeCodeOptions(
mcp_servers={"test": server},
disallowed_tools=["mcp__test__echo"], # Block echo tool
allowed_tools=["mcp__test__greet"], # But allow greet
allowed_tools=["mcp__test__greet"], # But allow greet
)
async with ClaudeSDKClient(options=options) as client:
await client.query("Use the echo tool to echo 'test' and use greet tool to greet 'Alice'")
await client.query(
"Use the echo tool to echo 'test' and use greet tool to greet 'Alice'"
)
async for message in client.receive_response():
pass # Just consume messages
# Check actual function executions
assert "echo" not in executions, "Disallowed echo tool was executed"
assert "greet" in executions, "Allowed greet tool was not executed"
@ -98,36 +97,38 @@ async def test_sdk_mcp_permission_enforcement():
async def test_sdk_mcp_multiple_tools():
"""Test that multiple SDK MCP tools can be called in sequence."""
executions = []
@tool("echo", "Echo back the input text", {"text": str})
async def echo_tool(args: dict[str, Any]) -> dict[str, Any]:
"""Echo back whatever text is provided."""
executions.append("echo")
return {"content": [{"type": "text", "text": f"Echo: {args['text']}"}]}
@tool("greet", "Greet a person by name", {"name": str})
async def greet_tool(args: dict[str, Any]) -> dict[str, Any]:
"""Greet someone by name."""
executions.append("greet")
return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]}
server = create_sdk_mcp_server(
name="multi",
version="1.0.0",
tools=[echo_tool, greet_tool],
)
options = ClaudeCodeOptions(
mcp_servers={"multi": server},
allowed_tools=["mcp__multi__echo", "mcp__multi__greet"],
)
async with ClaudeSDKClient(options=options) as client:
await client.query("Call mcp__multi__echo with text='test' and mcp__multi__greet with name='Bob'")
await client.query(
"Call mcp__multi__echo with text='test' and mcp__multi__greet with name='Bob'"
)
async for message in client.receive_response():
pass # Just consume messages
# Both tools should have been executed
assert "echo" in executions, "Echo tool was not executed"
assert "greet" in executions, "Greet tool was not executed"
@ -138,28 +139,28 @@ async def test_sdk_mcp_multiple_tools():
async def test_sdk_mcp_without_permissions():
"""Test SDK MCP tool behavior without explicit allowed_tools."""
executions = []
@tool("echo", "Echo back the input text", {"text": str})
async def echo_tool(args: dict[str, Any]) -> dict[str, Any]:
"""Echo back whatever text is provided."""
executions.append("echo")
return {"content": [{"type": "text", "text": f"Echo: {args['text']}"}]}
server = create_sdk_mcp_server(
name="noperm",
version="1.0.0",
tools=[echo_tool],
)
# No allowed_tools specified
options = ClaudeCodeOptions(
mcp_servers={"noperm": server},
)
async with ClaudeSDKClient(options=options) as client:
await client.query("Call the mcp__noperm__echo tool")
async for message in client.receive_response():
pass # Just consume messages
assert "echo" not in executions, "SDK MCP tool was executed"
assert "echo" not in executions, "SDK MCP tool was executed"

81
examples/mcp_calculator.py
View file

@ -20,17 +20,13 @@ from claude_code_sdk import (
# Define calculator tools using the @tool decorator
@tool("add", "Add two numbers", {"a": float, "b": float})
async def add_numbers(args: dict[str, Any]) -> dict[str, Any]:
"""Add two numbers together."""
result = args["a"] + args["b"]
return {
"content": [
{
"type": "text",
"text": f"{args['a']} + {args['b']} = {result}"
}
]
"content": [{"type": "text", "text": f"{args['a']} + {args['b']} = {result}"}]
}
@ -39,12 +35,7 @@ async def subtract_numbers(args: dict[str, Any]) -> dict[str, Any]:
"""Subtract b from a."""
result = args["a"] - args["b"]
return {
"content": [
{
"type": "text",
"text": f"{args['a']} - {args['b']} = {result}"
}
]
"content": [{"type": "text", "text": f"{args['a']} - {args['b']} = {result}"}]
}
@ -53,12 +44,7 @@ async def multiply_numbers(args: dict[str, Any]) -> dict[str, Any]:
"""Multiply two numbers."""
result = args["a"] * args["b"]
return {
"content": [
{
"type": "text",
"text": f"{args['a']} × {args['b']} = {result}"
}
]
"content": [{"type": "text", "text": f"{args['a']} × {args['b']} = {result}"}]
}
@ -68,22 +54,14 @@ async def divide_numbers(args: dict[str, Any]) -> dict[str, Any]:
if args["b"] == 0:
return {
"content": [
{
"type": "text",
"text": "Error: Division by zero is not allowed"
}
{"type": "text", "text": "Error: Division by zero is not allowed"}
],
"is_error": True
"is_error": True,
}
result = args["a"] / args["b"]
return {
"content": [
{
"type": "text",
"text": f"{args['a']} ÷ {args['b']} = {result}"
}
]
"content": [{"type": "text", "text": f"{args['a']} ÷ {args['b']} = {result}"}]
}
@ -96,22 +74,16 @@ async def square_root(args: dict[str, Any]) -> dict[str, Any]:
"content": [
{
"type": "text",
"text": f"Error: Cannot calculate square root of negative number {n}"
"text": f"Error: Cannot calculate square root of negative number {n}",
}
],
"is_error": True
"is_error": True,
}
import math
result = math.sqrt(n)
return {
"content": [
{
"type": "text",
"text": f"√{n} = {result}"
}
]
}
return {"content": [{"type": "text", "text": f"√{n} = {result}"}]}
@tool("power", "Raise a number to a power", {"base": float, "exponent": float})
@ -120,10 +92,7 @@ async def power(args: dict[str, Any]) -> dict[str, Any]:
result = args["base"] ** args["exponent"]
return {
"content": [
{
"type": "text",
"text": f"{args['base']}^{args['exponent']} = {result}"
}
{"type": "text", "text": f"{args['base']}^{args['exponent']} = {result}"}
]
}
@ -139,13 +108,15 @@ def display_message(msg):
ToolUseBlock,
UserMessage,
)
if isinstance(msg, UserMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"User: {block.text}")
elif isinstance(block, ToolResultBlock):
print(f"Tool Result: {block.content[:100] if block.content else 'None'}...")
print(
f"Tool Result: {block.content[:100] if block.content else 'None'}..."
)
elif isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
@ -178,8 +149,8 @@ async def main():
multiply_numbers,
divide_numbers,
square_root,
power
]
power,
],
)
# Configure Claude to use the calculator server with allowed tools
@ -188,12 +159,12 @@ async def main():
mcp_servers={"calc": calculator},
allowed_tools=[
"mcp__calc__add",
"mcp__calc__subtract",
"mcp__calc__subtract",
"mcp__calc__multiply",
"mcp__calc__divide",
"mcp__calc__sqrt",
"mcp__calc__power"
]
"mcp__calc__power",
],
)
# Example prompts to demonstrate calculator usage
@ -203,17 +174,17 @@ async def main():
"What is 100 divided by 7?",
"Calculate the square root of 144",
"What is 2 raised to the power of 8?",
"Calculate (12 + 8) * 3 - 10" # Complex calculation
"Calculate (12 + 8) * 3 - 10", # Complex calculation
]
for prompt in prompts:
print(f"\n{'='*50}")
print(f"\n{'=' * 50}")
print(f"Prompt: {prompt}")
print(f"{'='*50}")
print(f"{'=' * 50}")
async with ClaudeSDKClient(options=options) as client:
await client.query(prompt)
async for message in client.receive_response():
display_message(message)

59
src/claude_code_sdk/client.py
View file

@ -39,57 +39,16 @@ class ClaudeSDKClient:
- When all inputs are known upfront
- Stateless operations
Example - Interactive conversation:
```python
# Automatically connects with empty stream for interactive use
async with ClaudeSDKClient() as client:
# Send initial message
await client.query("Let's solve a math problem step by step")
See examples/streaming_mode.py for full examples of ClaudeSDKClient in
different scenarios.
# Receive and process response
async for message in client.receive_messages():
if "ready" in str(message.content).lower():
break
# Send follow-up based on response
await client.query("What's 15% of 80?")
# Continue conversation...
# Automatically disconnects
```
Example - With interrupt:
```python
async with ClaudeSDKClient() as client:
# Start a long task
await client.query("Count to 1000")
# Interrupt after 2 seconds
await anyio.sleep(2)
await client.interrupt()
# Send new instruction
await client.query("Never mind, what's 2+2?")
```
Example - Manual connection:
```python
client = ClaudeSDKClient()
# Connect with initial message stream
async def message_stream():
yield {"type": "user", "message": {"role": "user", "content": "Hello"}}
await client.connect(message_stream())
# Send additional messages dynamically
await client.query("What's the weather?")
async for message in client.receive_messages():
print(message)
await client.disconnect()
```
Caveat: As of v0.0.20, you cannot use a ClaudeSDKClient instance across
different async runtime contexts (e.g., different trio nurseries or asyncio
task groups). The client internally maintains a persistent anyio task group
for reading messages that remains active from connect() until disconnect().
This means you must complete all operations with the client within the same
async context where it was connected. Ideally, this limitation should not
exist.
"""
def __init__(self, options: ClaudeCodeOptions | None = None):