Overview
The Claude Agent SDK allows you to create custom tools that run in-process within your Python application. These tools extend Claude’s capabilities with your own functions and logic.
SDK MCP servers run in-process within your application, providing better performance than external MCP servers that require separate processes.
Benefits of SDK MCP Servers
Better Performance No IPC overhead - tools run directly in your process
Simpler Deployment Single process - no need to manage separate server processes
Easier Debugging Debug tools alongside your application code
State Access Direct access to your application’s variables and state
Use the @tool decorator to define a custom tool:
from claude_agent_sdk import tool
from typing import Any
@tool (
name = "greet" ,
description = "Greet a user by name" ,
input_schema = { "name" : str }
)
async def greet ( args : dict[ str , Any]) -> dict[ str , Any]:
"""Greet a user."""
return {
"content" : [
{ "type" : "text" , "text" : f "Hello, { args[ 'name' ] } !" }
]
}
Tool functions must be async (defined with async def) and return a dictionary with a content key containing the response.
Every tool needs three components:
Name
A unique identifier that Claude uses to reference the tool: @tool (
name = "calculate_sum" , # Unique identifier
# ...
)
Description
Human-readable description that helps Claude understand when to use the tool: @tool (
name = "calculate_sum" ,
description = "Add two numbers together and return the sum" , # Clear description
# ...
)
Input Schema
Schema defining the tool’s input parameters: @tool (
name = "calculate_sum" ,
description = "Add two numbers together and return the sum" ,
input_schema = { "a" : float , "b" : float } # Parameter types
)
Complete Calculator Example
Here’s a complete example from the SDK examples:
import asyncio
from typing import Any
from claude_agent_sdk import (
tool,
create_sdk_mcp_server,
ClaudeSDKClient,
ClaudeAgentOptions
)
# Define calculator tools
@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 } " }]
}
@tool ( "subtract" , "Subtract one number from another" , { "a" : float , "b" : float })
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 } " }]
}
@tool ( "multiply" , "Multiply two numbers" , { "a" : float , "b" : float })
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 } " }]
}
@tool ( "divide" , "Divide one number by another" , { "a" : float , "b" : float })
async def divide_numbers ( args : dict[ str , Any]) -> dict[ str , Any]:
"""Divide a by b."""
if args[ "b" ] == 0 :
return {
"content" : [
{ "type" : "text" , "text" : "Error: Division by zero is not allowed" }
],
"is_error" : True
}
result = args[ "a" ] / args[ "b" ]
return {
"content" : [{ "type" : "text" , "text" : f " { args[ 'a' ] } ÷ { args[ 'b' ] } = { result } " }]
}
# Create the MCP server
calculator = create_sdk_mcp_server(
name = "calculator" ,
version = "2.0.0" ,
tools = [add_numbers, subtract_numbers, multiply_numbers, divide_numbers]
)
# Configure Claude to use the calculator
options = ClaudeAgentOptions(
mcp_servers = { "calc" : calculator},
allowed_tools = [
"mcp__calc__add" ,
"mcp__calc__subtract" ,
"mcp__calc__multiply" ,
"mcp__calc__divide"
]
)
async def main ():
async with ClaudeSDKClient( options = options) as client:
await client.query( "Calculate 15 + 27" )
async for message in client.receive_response():
print (message)
asyncio.run(main())
Error Handling
Handle errors gracefully in your tools:
Division by Zero
Validation
Try-Except
@tool ( "divide" , "Divide two numbers" , { "a" : float , "b" : float })
async def divide ( args : dict[ str , Any]) -> dict[ str , Any]:
if args[ "b" ] == 0 :
return {
"content" : [
{ "type" : "text" , "text" : "Error: Division by zero" }
],
"is_error" : True # Mark as error
}
result = args[ "a" ] / args[ "b" ]
return {
"content" : [{ "type" : "text" , "text" : f "Result: { result } " }]
}
Accessing Application State
Tools can directly access your application’s state:
from claude_agent_sdk import tool, create_sdk_mcp_server
from typing import Any
# Application state
class DataStore :
def __init__ ( self ):
self .items = []
self .counter = 0
store = DataStore()
# Tools that access the store
@tool ( "add_item" , "Add item to store" , { "item" : str })
async def add_item ( args : dict[ str , Any]) -> dict[ str , Any]:
store.items.append(args[ "item" ])
store.counter += 1
return {
"content" : [
{ "type" : "text" , "text" : f "Added: { args[ 'item' ] } (total: { store.counter } )" }
]
}
@tool ( "list_items" , "List all items in store" , {})
async def list_items ( args : dict[ str , Any]) -> dict[ str , Any]:
if not store.items:
return { "content" : [{ "type" : "text" , "text" : "Store is empty" }]}
items_text = " \n " .join( f " { i + 1 } . { item } " for i, item in enumerate (store.items))
return {
"content" : [{ "type" : "text" , "text" : f "Items in store: \n { items_text } " }]
}
@tool ( "clear_store" , "Clear all items" , {})
async def clear_store ( args : dict[ str , Any]) -> dict[ str , Any]:
count = len (store.items)
store.items.clear()
store.counter = 0
return {
"content" : [{ "type" : "text" , "text" : f "Cleared { count } items" }]
}
# Create server with all tools
server = create_sdk_mcp_server(
name = "datastore" ,
tools = [add_item, list_items, clear_store]
)
Multiple Parameter Types
@tool (
"format_message" ,
"Format a message with various options" ,
{
"text" : str ,
"uppercase" : bool ,
"repeat_count" : int ,
"prefix" : str
}
)
async def format_message ( args : dict[ str , Any]) -> dict[ str , Any]:
text = args[ "text" ]
if args.get( "uppercase" , False ):
text = text.upper()
repeat = args.get( "repeat_count" , 1 )
prefix = args.get( "prefix" , "" )
result = " \n " .join([ f " { prefix }{ text } " for _ in range (repeat)])
return { "content" : [{ "type" : "text" , "text" : result}]}
JSON Schema
For complex validation, use full JSON Schema:
@tool (
"create_user" ,
"Create a new user" ,
{
"type" : "object" ,
"properties" : {
"name" : { "type" : "string" , "minLength" : 1 },
"age" : { "type" : "integer" , "minimum" : 0 , "maximum" : 150 },
"email" : { "type" : "string" , "format" : "email" }
},
"required" : [ "name" , "email" ]
}
)
async def create_user ( args : dict[ str , Any]) -> dict[ str , Any]:
user = {
"name" : args[ "name" ],
"email" : args[ "email" ],
"age" : args.get( "age" , 0 )
}
return {
"content" : [
{ "type" : "text" , "text" : f "Created user: { user[ 'name' ] } ( { user[ 'email' ] } )" }
]
}
When using MCP servers, tools are prefixed with the server name:
# Create server named "calc"
calculator = create_sdk_mcp_server(
name = "calc" ,
tools = [add_numbers]
)
# Tool "add" becomes "mcp__calc__add" in allowed_tools
options = ClaudeAgentOptions(
mcp_servers = { "calc" : calculator},
allowed_tools = [ "mcp__calc__add" ] # Format: mcp__{server}__{tool}
)
Always use the mcp__{server}__{tool} format in allowed_tools when pre-approving MCP server tools.
Using with query()
Tools work with both ClaudeSDKClient and query():
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, tool, create_sdk_mcp_server
from typing import Any
@tool ( "hello" , "Say hello" , { "name" : str })
async def say_hello ( args : dict[ str , Any]) -> dict[ str , Any]:
return { "content" : [{ "type" : "text" , "text" : f "Hello, { args[ 'name' ] } !" }]}
server = create_sdk_mcp_server( name = "greeter" , tools = [say_hello])
options = ClaudeAgentOptions(
mcp_servers = { "greeter" : server},
allowed_tools = [ "mcp__greeter__hello" ]
)
async def main ():
async for message in query(
prompt = "Greet Alice using your tools" ,
options = options
):
print (message)
asyncio.run(main())
Multiple MCP Servers
You can register multiple MCP servers:
from claude_agent_sdk import ClaudeAgentOptions
# Create multiple servers
calculator = create_sdk_mcp_server( name = "calc" , tools = [add, subtract])
data_store = create_sdk_mcp_server( name = "store" , tools = [add_item, list_items])
weather = create_sdk_mcp_server( name = "weather" , tools = [get_weather])
# Register all servers
options = ClaudeAgentOptions(
mcp_servers = {
"calc" : calculator,
"store" : data_store,
"weather" : weather
},
allowed_tools = [
# Calculator tools
"mcp__calc__add" ,
"mcp__calc__subtract" ,
# Data store tools
"mcp__store__add_item" ,
"mcp__store__list_items" ,
# Weather tools
"mcp__weather__get_weather"
]
)
Best Practices
Write clear, specific descriptions that help Claude understand when to use each tool: # Good
description = "Add two numbers together and return the sum"
# Bad
description = "Math operation"
Provide helpful error messages that guide Claude: return {
"content" : [{ "type" : "text" , "text" : "Error: File not found. Please check the path." }],
"is_error" : True
}
Always define tools as async functions, even if they don’t use await: async def my_tool ( args ): # async def, not def
return { ... }
Complete Working Example
Here’s a complete, runnable example:
#!/usr/bin/env python3
import asyncio
from typing import Any
from claude_agent_sdk import (
tool,
create_sdk_mcp_server,
ClaudeSDKClient,
ClaudeAgentOptions,
AssistantMessage,
TextBlock,
ToolUseBlock
)
# Define tools
@tool ( "add" , "Add two numbers" , { "a" : float , "b" : float })
async def add_numbers ( args : dict[ str , Any]) -> dict[ str , Any]:
result = args[ "a" ] + args[ "b" ]
return {
"content" : [{ "type" : "text" , "text" : f " { args[ 'a' ] } + { args[ 'b' ] } = { result } " }]
}
@tool ( "sqrt" , "Calculate square root" , { "n" : float })
async def square_root ( args : dict[ str , Any]) -> dict[ str , Any]:
n = args[ "n" ]
if n < 0 :
return {
"content" : [{ "type" : "text" , "text" : f "Error: Cannot calculate √ { n } " }],
"is_error" : True
}
import math
result = math.sqrt(n)
return { "content" : [{ "type" : "text" , "text" : f "√ { n } = { result } " }]}
# Create server
calculator = create_sdk_mcp_server(
name = "calculator" ,
version = "1.0.0" ,
tools = [add_numbers, square_root]
)
# Configure options
options = ClaudeAgentOptions(
mcp_servers = { "calc" : calculator},
allowed_tools = [ "mcp__calc__add" , "mcp__calc__sqrt" ]
)
async def main ():
async with ClaudeSDKClient( options = options) as client:
# Test the calculator
await client.query( "Calculate the square root of 144, then add 5 to it" )
async for msg in client.receive_response():
if isinstance (msg, AssistantMessage):
for block in msg.content:
if isinstance (block, TextBlock):
print ( f "Claude: { block.text } " )
elif isinstance (block, ToolUseBlock):
print ( f "Using tool: { block.name } " )
if __name__ == "__main__" :
asyncio.run(main())
Next Steps
Hooks Implement hooks to control tool execution
Permissions Learn about tool permission management
MCP Servers Explore external MCP servers
Examples See the full calculator example