Claude Agent SDK Python: The Revolutionary Framework That's Transforming AI Development with Custom Tools and Bidirectional Communication
A deep dive into the Claude Agent SDK Python: learn installation, advanced usage, custom tool creation, hooks, error handling, and real-world AI agent development with practical code examples.
Introduction: The Future of AI Agent Development
The landscape of AI development is rapidly evolving, and Anthropic's Claude Agent SDK for Python represents a significant leap forward in how developers can build and integrate AI agents into their applications. This powerful SDK, recently released and gaining massive traction with over 1,900 GitHub stars, provides developers with unprecedented control over Claude's capabilities through custom tools, bidirectional communication, and sophisticated hook systems.
Unlike traditional AI integrations that rely on simple request-response patterns, the Claude Agent SDK Python enables developers to create truly interactive, tool-enabled AI agents that can execute code, manipulate files, and integrate seamlessly with existing Python applications. Whether you're building automation tools, development assistants, or complex AI-powered workflows, this SDK provides the foundation for next-generation AI applications.
What Makes Claude Agent SDK Python Revolutionary?
1. Bidirectional Communication
The SDK implements a sophisticated control protocol that enables real-time, bidirectional communication between your Python application and Claude. This means Claude can not only respond to queries but also request permissions, provide status updates, and maintain persistent conversations.
2. Custom Tools as In-Process MCP Servers
One of the most groundbreaking features is the ability to define custom tools as Python functions that run directly within your application process. This eliminates the complexity of managing separate MCP (Model Context Protocol) server processes while providing better performance and easier debugging.
3. Advanced Hook System
The hook system allows you to inject custom logic at specific points in Claude's execution flow, enabling automated validation, custom processing, and deterministic behavior control.
Installation and Prerequisites
Getting started with Claude Agent SDK Python requires a few prerequisites and a straightforward installation process.
Prerequisites
- Python 3.10+ - The SDK requires modern Python features
- Node.js - Required for the Claude Code CLI
- Claude Code CLI - The underlying engine that powers the SDK
Step-by-Step Installation
# Install the Python SDK
pip install claude-agent-sdk
# Install Claude Code CLI globally
npm install -g @anthropic-ai/claude-code
# Verify installation
claude-code --versionQuick Start: Your First Claude Agent
Let's start with a simple example to understand the basic usage pattern:
import anyio
from claude_agent_sdk import query
async def main():
# Simple query example
async for message in query(prompt="What is 2 + 2?"):
print(message)
anyio.run(main())This basic example demonstrates the streaming nature of the SDK - responses are delivered as an async iterator, allowing for real-time processing of Claude's output.
Advanced Usage with ClaudeAgentOptions
For more sophisticated applications, you'll want to configure Claude's behavior using ClaudeAgentOptions:
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock
async def advanced_query_example():
# Configure Claude's behavior
options = ClaudeAgentOptions(
system_prompt="You are a helpful Python development assistant",
max_turns=5,
allowed_tools=["Read", "Write", "Bash"],
permission_mode='acceptEdits', # Auto-accept file edits
cwd="/path/to/your/project" # Set working directory
)
async for message in query(
prompt="Create a Python script that calculates fibonacci numbers",
options=options
):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude says: {block.text}")
anyio.run(advanced_query_example())Building Custom Tools: The Game-Changer
The ability to create custom tools as in-process MCP servers is perhaps the most powerful feature of the Claude Agent SDK Python. Here's how to implement your own tools:
Creating a Simple Custom Tool
from claude_agent_sdk import (
tool,
create_sdk_mcp_server,
ClaudeAgentOptions,
ClaudeSDKClient
)
# Define a custom tool using the @tool decorator
@tool("calculate_fibonacci", "Calculate fibonacci number at position n", {"n": int})
async def calculate_fibonacci(args):
n = args['n']
if n <= 1:
result = n
else:
a, b = 0, 1
for _ in range(2, n + 1):
a, b = b, a + b
result = b
return {
"content": [
{
"type": "text",
"text": f"The {n}th Fibonacci number is {result}"
}
]
}
# Create an SDK MCP server with your custom tools
math_server = create_sdk_mcp_server(
name="math-tools",
version="1.0.0",
tools=[calculate_fibonacci]
)
async def use_custom_tools():
options = ClaudeAgentOptions(
mcp_servers={"math": math_server},
allowed_tools=["mcp__math__calculate_fibonacci"]
)
async with ClaudeSDKClient(options=options) as client:
await client.query("Calculate the 10th Fibonacci number")
# Process the response
async for msg in client.receive_response():
print(msg)
anyio.run(use_custom_tools())Advanced Custom Tool with Database Integration
import sqlite3
from claude_agent_sdk import tool, create_sdk_mcp_server
@tool(
"query_database",
"Execute SQL queries on the application database",
{"query": str, "params": list}
)
async def query_database(args):
query = args['query']
params = args.get('params', [])
try:
conn = sqlite3.connect('app.db')
cursor = conn.cursor()
cursor.execute(query, params)
if query.strip().upper().startswith('SELECT'):
results = cursor.fetchall()
columns = [description[0] for description in cursor.description]
# Format results as a table
table_data = []
for row in results:
table_data.append(dict(zip(columns, row)))
return {
"content": [
{
"type": "text",
"text": f"Query executed successfully. Results:\n{table_data}"
}
]
}
else:
conn.commit()
return {
"content": [
{
"type": "text",
"text": f"Query executed successfully. Rows affected: {cursor.rowcount}"
}
]
}
except Exception as e:
return {
"content": [
{
"type": "text",
"text": f"Database error: {str(e)}"
}
]
}
finally:
conn.close()
# Create database tools server
db_server = create_sdk_mcp_server(
name="database-tools",
version="1.0.0",
tools=[query_database]
)Implementing Hooks for Advanced Control
Hooks provide a powerful way to inject custom logic into Claude's execution flow. Here's how to implement security and validation hooks:
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher
async def security_check_hook(input_data, tool_use_id, context):
"""Hook to validate bash commands before execution"""
tool_name = input_data["tool_name"]
tool_input = input_data["tool_input"]
if tool_name != "Bash":
return {} # Not a bash command, allow it
command = tool_input.get("command", "")
# Define dangerous patterns
dangerous_patterns = [
"rm -rf",
"sudo",
"chmod 777",
"wget",
"curl"
]
for pattern in dangerous_patterns:
if pattern in command:
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": f"Command contains dangerous pattern: {pattern}",
}
}
return {} # Allow the command
async def logging_hook(input_data, tool_use_id, context):
"""Hook to log all tool usage"""
tool_name = input_data["tool_name"]
print(f"Tool used: {tool_name} at {context.get('timestamp', 'unknown time')}")
return {}
async def use_hooks_example():
options = ClaudeAgentOptions(
allowed_tools=["Bash", "Read", "Write"],
hooks={
"PreToolUse": [
HookMatcher(matcher="Bash", hooks=[security_check_hook]),
HookMatcher(matcher="*", hooks=[logging_hook]), # Log all tools
],
}
)
async with ClaudeSDKClient(options=options) as client:
# Test dangerous command (will be blocked)
await client.query("Run: rm -rf /important/data")
async for msg in client.receive_response():
print(msg)
print("\n" + "="*50 + "\n")
# Test safe command (will execute)
await client.query("Run: echo 'Hello, secure world!'")
async for msg in client.receive_response():
print(msg)
anyio.run(use_hooks_example())Error Handling and Best Practices
Robust error handling is crucial for production applications. The SDK provides comprehensive error types:
from claude_agent_sdk import (
ClaudeSDKError, # Base error
CLINotFoundError, # Claude Code not installed
CLIConnectionError, # Connection issues
ProcessError, # Process failed
CLIJSONDecodeError, # JSON parsing issues
query
)
async def robust_query_example():
try:
async for message in query(prompt="Analyze this complex dataset"):
# Process message
print(message)
except CLINotFoundError:
print("Error: Claude Code CLI not found. Please install with:")
print("npm install -g @anthropic-ai/claude-code")
except CLIConnectionError as e:
print(f"Connection error: {e}")
print("Check your network connection and API credentials")
except ProcessError as e:
print(f"Process failed with exit code: {e.exit_code}")
print(f"Error output: {e.stderr}")
except CLIJSONDecodeError as e:
print(f"Failed to parse Claude's response: {e}")
print("This might indicate a version mismatch")
except ClaudeSDKError as e:
print(f"General SDK error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
anyio.run(robust_query_example())Real-World Application: Building an AI Code Review Assistant
Let's put everything together in a practical example - an AI-powered code review assistant:
import os
import subprocess
from pathlib import Path
from claude_agent_sdk import (
tool, create_sdk_mcp_server, ClaudeAgentOptions,
ClaudeSDKClient, HookMatcher
)
@tool("run_linter", "Run code linting on specified files", {"files": list})
async def run_linter(args):
files = args['files']
results = []
for file_path in files:
if file_path.endswith('.py'):
try:
result = subprocess.run(
['flake8', file_path],
capture_output=True,
text=True
)
if result.returncode == 0:
results.append(f"✅ {file_path}: No issues found")
else:
results.append(f"❌ {file_path}:\n{result.stdout}")
except FileNotFoundError:
results.append(f"⚠️ {file_path}: flake8 not installed")
return {
"content": [
{
"type": "text",
"text": "\n".join(results)
}
]
}
@tool("get_git_diff", "Get git diff for code review", {})
async def get_git_diff(args):
try:
result = subprocess.run(
['git', 'diff', '--cached'],
capture_output=True,
text=True
)
return {
"content": [
{
"type": "text",
"text": f"Git diff:\n```\n{result.stdout}\n```"
}
]
}
except Exception as e:
return {
"content": [
{
"type": "text",
"text": f"Error getting git diff: {e}"
}
]
}
async def code_review_hook(input_data, tool_use_id, context):
"""Log all code review activities"""
tool_name = input_data["tool_name"]
timestamp = context.get('timestamp', 'unknown')
print(f"[{timestamp}] Code review tool used: {tool_name}")
return {}
async def ai_code_reviewer():
# Create code review tools server
review_server = create_sdk_mcp_server(
name="code-review-tools",
version="1.0.0",
tools=[run_linter, get_git_diff]
)
options = ClaudeAgentOptions(
system_prompt="""
You are an expert code reviewer. Your job is to:
1. Analyze code changes using git diff
2. Run linting tools to check code quality
3. Provide constructive feedback on code structure, performance, and best practices
4. Suggest improvements and identify potential bugs
Always be thorough but constructive in your reviews.
""",
mcp_servers={"review": review_server},
allowed_tools=[
"mcp__review__run_linter",
"mcp__review__get_git_diff",
"Read",
"Write"
],
hooks={
"PreToolUse": [
HookMatcher(matcher="*", hooks=[code_review_hook])
]
},
cwd=os.getcwd()
)
async with ClaudeSDKClient(options=options) as client:
await client.query(
"Please review the staged changes in this repository. "
"Check the git diff, run linting tools, and provide a comprehensive code review."
)
async for msg in client.receive_response():
print(msg)
if __name__ == "__main__":
anyio.run(ai_code_reviewer())Performance Optimization and Best Practices
1. Connection Management
Always use the async context manager pattern with ClaudeSDKClient to ensure proper cleanup:
async with ClaudeSDKClient(options=options) as client:
# Your code here
pass
# Connection automatically cleaned up2. Tool Organization
Group related tools into logical MCP servers:
# Good: Organized by functionality
file_server = create_sdk_mcp_server("file-ops", tools=[read_file, write_file])
db_server = create_sdk_mcp_server("database", tools=[query_db, update_db])
# Avoid: All tools in one server
monolith_server = create_sdk_mcp_server("everything", tools=[...50_tools...])3. Error Recovery
Implement graceful degradation when tools fail:
@tool("safe_operation", "Perform operation with fallback", {"data": str})
async def safe_operation(args):
try:
# Primary operation
result = await primary_operation(args['data'])
return {"content": [{"type": "text", "text": result}]}
except Exception as e:
# Fallback operation
fallback_result = await fallback_operation(args['data'])
return {
"content": [{
"type": "text",
"text": f"Primary operation failed ({e}), used fallback: {fallback_result}"
}]
}Migration from External MCP Servers
If you're currently using external MCP servers, migrating to SDK MCP servers offers significant benefits:
Before: External MCP Server
# External server configuration
options = ClaudeAgentOptions(
mcp_servers={
"calculator": {
"type": "stdio",
"command": "python",
"args": ["-m", "calculator_server"]
}
}
)After: SDK MCP Server
# In-process server
from my_tools import add, subtract, multiply, divide
calculator = create_sdk_mcp_server(
name="calculator",
tools=[add, subtract, multiply, divide]
)
options = ClaudeAgentOptions(
mcp_servers={"calculator": calculator}
)Future Developments and Roadmap
The Claude Agent SDK Python is actively developed with exciting features on the horizon:
- Enhanced Control Protocol: More sophisticated bidirectional communication patterns
- Advanced Hook Types: New hook points for finer-grained control
- Performance Improvements: Optimized message passing and reduced latency
- Extended Tool Ecosystem: Pre-built tool libraries for common use cases
Conclusion: Embracing the Future of AI Development
The Claude Agent SDK Python represents a paradigm shift in AI application development. By providing developers with powerful tools for creating custom functionality, implementing sophisticated control flows, and maintaining bidirectional communication with Claude, it opens up possibilities that were previously impossible or extremely complex to implement.
Whether you're building development tools, automation systems, or complex AI-powered applications, the Claude Agent SDK Python provides the foundation you need to create truly intelligent, interactive systems. The combination of custom tools, hooks, and bidirectional communication creates a platform where AI agents can become true partners in your development workflow.
As the AI landscape continues to evolve, frameworks like the Claude Agent SDK Python will become increasingly important for developers who want to stay at the forefront of AI application development. Start experimenting with the SDK today, and discover how it can transform your approach to building AI-powered applications.
For more expert insights and tutorials on AI and automation, visit us at decisioncrafters.com.