Custom Tools (In-Process MCP Servers)

def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any]
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]:
    """
    Decorator for creating custom tools.

    Creates a tool that can be used with SDK MCP servers. The tool runs
    in-process within your Python application, providing better performance
    than external MCP servers.

    Args:
        name: Unique identifier for the tool. This is what Claude will use
            to reference the tool in function calls.
        description: Human-readable description of what the tool does.
            This helps Claude understand when to use the tool.
        input_schema: Schema defining the tool's input parameters. Can be:
            - A dictionary mapping parameter names to types (e.g., {"text": str})
            - A TypedDict class for more complex schemas
            - A JSON Schema dictionary for full validation

    Returns:
        A decorator function that wraps the tool implementation and returns
        an SdkMcpTool instance ready for use with create_sdk_mcp_server().

    Notes:
        - The tool function must be async (defined with async def)
        - The function receives a single dict argument with the input parameters
        - The function should return a dict with a "content" key containing the response
        - Errors can be indicated by including "is_error": True in the response
    """

from claude_agent_sdk import tool

@tool("greet", "Greet a user by name", {"name": str})
async def greet(args):
    return {
        "content": [
            {"type": "text", "text": f"Hello, {args['name']}!"}
        ]
    }

@tool("add", "Add two numbers", {"a": float, "b": float})
async def add_numbers(args):
    result = args["a"] + args["b"]
    return {
        "content": [
            {"type": "text", "text": f"Result: {result}"}
        ]
    }

@tool("divide", "Divide two numbers", {"a": float, "b": float})
async def divide(args):
    if args["b"] == 0:
        return {
            "content": [{"type": "text", "text": "Error: Division by zero"}],
            "is_error": True
        }
    result = args["a"] / args["b"]
    return {
        "content": [{"type": "text", "text": f"Result: {result}"}]
    }

from typing import TypedDict

class SearchInput(TypedDict):
    query: str
    limit: int
    filters: dict[str, str]

@tool("search", "Search database", SearchInput)
async def search_database(args):
    query = args["query"]
    limit = args.get("limit", 10)
    filters = args.get("filters", {})

    # Perform search
    results = await db.search(query, limit, filters)

    return {
        "content": [
            {"type": "text", "text": f"Found {len(results)} results"}
        ]
    }

schema = {
    "type": "object",
    "properties": {
        "text": {"type": "string", "minLength": 1},
        "count": {"type": "integer", "minimum": 0, "maximum": 100}
    },
    "required": ["text"]
}

@tool("repeat", "Repeat text multiple times", schema)
async def repeat_text(args):
    text = args["text"]
    count = args.get("count", 1)
    return {
        "content": [
            {"type": "text", "text": text * count}
        ]
    }

def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig:
    """
    Create an in-process MCP server that runs within your Python application.

    Unlike external MCP servers that run as separate processes, SDK MCP servers
    run directly in your application's process. This provides:
    - Better performance (no IPC overhead)
    - Simpler deployment (single process)
    - Easier debugging (same process)
    - Direct access to your application's state

    Args:
        name: Unique identifier for the server. This name is used to reference
            the server in the mcp_servers configuration.
        version: Server version string. Defaults to "1.0.0". This is for
            informational purposes and doesn't affect functionality.
        tools: List of SdkMcpTool instances created with the @tool decorator.
            These are the functions that Claude can call through this server.
            If None or empty, the server will have no tools (rarely useful).

    Returns:
        McpSdkServerConfig: A configuration object that can be passed to
        ClaudeAgentOptions.mcp_servers. This config contains the server
        instance and metadata needed for the SDK to route tool calls.

    Notes:
        - The server runs in the same process as your Python application
        - Tools have direct access to your application's variables and state
        - No subprocess or IPC overhead for tool calls
        - Server lifecycle is managed automatically by the SDK
    """

from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions

@tool("add", "Add numbers", {"a": float, "b": float})
async def add(args):
    return {"content": [{"type": "text", "text": f"Sum: {args['a'] + args['b']}"}]}

@tool("multiply", "Multiply numbers", {"a": float, "b": float})
async def multiply(args):
    return {"content": [{"type": "text", "text": f"Product: {args['a'] * args['b']}"}]}

# Create server
calculator = create_sdk_mcp_server(
    name="calculator",
    version="2.0.0",
    tools=[add, multiply]
)

# Use with Claude
options = ClaudeAgentOptions(
    mcp_servers={"calc": calculator},
    allowed_tools=["add", "multiply"]
)

from claude_agent_sdk import tool, create_sdk_mcp_server

class DataStore:
    def __init__(self):
        self.items = []

    def add_item(self, item):
        self.items.append(item)

    def get_items(self):
        return self.items

# Create store instance
store = DataStore()

# Tools have access to store
@tool("add_item", "Add item to store", {"item": str})
async def add_item(args):
    store.add_item(args["item"])
    return {"content": [{"type": "text", "text": f"Added: {args['item']}"}]}

@tool("list_items", "List all items", {})
async def list_items(args):
    items = store.get_items()
    return {"content": [{"type": "text", "text": f"Items: {items}"}]}

# Create server
server = create_sdk_mcp_server("store", tools=[add_item, list_items])

@dataclass
class SdkMcpTool(Generic[T]):
    """Custom tool definition."""

    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]

from claude_agent_sdk import SdkMcpTool

async def my_handler(args):
    return {"content": [{"type": "text", "text": "Result"}]}

tool_instance = SdkMcpTool(
    name="my_tool",
    description="Does something",
    input_schema={"param": str},
    handler=my_handler
)

from claude_agent_sdk import (
    tool, create_sdk_mcp_server, query,
    ClaudeAgentOptions, AssistantMessage, TextBlock
)
import anyio

@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
    result = args["a"] + args["b"]
    return {"content": [{"type": "text", "text": f"{result}"}]}

@tool("subtract", "Subtract two numbers", {"a": float, "b": float})
async def subtract(args):
    result = args["a"] - args["b"]
    return {"content": [{"type": "text", "text": f"{result}"}]}

@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply(args):
    result = args["a"] * args["b"]
    return {"content": [{"type": "text", "text": f"{result}"}]}

@tool("divide", "Divide two numbers", {"a": float, "b": float})
async def divide(args):
    if args["b"] == 0:
        return {
            "content": [{"type": "text", "text": "Error: Division by zero"}],
            "is_error": True
        }
    result = args["a"] / args["b"]
    return {"content": [{"type": "text", "text": f"{result}"}]}

# Create calculator server
calc_server = create_sdk_mcp_server(
    name="calculator",
    version="1.0.0",
    tools=[add, subtract, multiply, divide]
)

# Use calculator
async def main():
    options = ClaudeAgentOptions(
        mcp_servers={"calc": calc_server},
        allowed_tools=["add", "subtract", "multiply", "divide"]
    )

    async for msg in query(prompt="Calculate (42 * 17) + (100 / 4)", options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)

from claude_agent_sdk import tool, create_sdk_mcp_server
from typing import TypedDict, Literal

class QueryInput(TypedDict):
    sql: str
    params: dict[str, Any]

class InsertInput(TypedDict):
    table: str
    data: dict[str, Any]

class Database:
    def __init__(self):
        self.tables = {}

    async def query(self, sql: str, params: dict):
        # Execute query
        pass

    async def insert(self, table: str, data: dict):
        # Insert data
        if table not in self.tables:
            self.tables[table] = []
        self.tables[table].append(data)

# Create database instance
db = Database()

@tool("db_query", "Execute SQL query", QueryInput)
async def db_query(args):
    result = await db.query(args["sql"], args["params"])
    return {"content": [{"type": "text", "text": f"Query result: {result}"}]}

@tool("db_insert", "Insert data into table", InsertInput)
async def db_insert(args):
    await db.insert(args["table"], args["data"])
    return {"content": [{"type": "text", "text": "Data inserted successfully"}]}

# Create server
db_server = create_sdk_mcp_server(
    name="database",
    version="1.0.0",
    tools=[db_query, db_insert]
)

from claude_agent_sdk import tool, create_sdk_mcp_server
import hashlib

@tool("hash_file", "Calculate file hash", {"path": str, "algorithm": str})
async def hash_file(args):
    path = args["path"]
    algorithm = args.get("algorithm", "sha256")

    try:
        hasher = hashlib.new(algorithm)
        with open(path, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hasher.update(chunk)
        return {
            "content": [{"type": "text", "text": f"{algorithm}: {hasher.hexdigest()}"}]
        }
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {e}"}],
            "is_error": True
        }

@tool("file_stats", "Get file statistics", {"path": str})
async def file_stats(args):
    import os
    try:
        stats = os.stat(args["path"])
        return {
            "content": [{
                "type": "text",
                "text": f"Size: {stats.st_size} bytes, Modified: {stats.st_mtime}"
            }]
        }
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {e}"}],
            "is_error": True
        }

# Create server
file_server = create_sdk_mcp_server(
    name="file_tools",
    version="1.0.0",
    tools=[hash_file, file_stats]
)

from claude_agent_sdk import tool, create_sdk_mcp_server
import httpx

class APIClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.api_key = api_key

    async def get(self, endpoint: str):
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}{endpoint}",
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            return response.json()

    async def post(self, endpoint: str, data: dict):
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}{endpoint}",
                json=data,
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            return response.json()

# Create API client
api = APIClient("https://api.example.com", "secret-key")

@tool("api_get", "Fetch data from API", {"endpoint": str})
async def api_get(args):
    try:
        data = await api.get(args["endpoint"])
        return {"content": [{"type": "text", "text": str(data)}]}
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"API Error: {e}"}],
            "is_error": True
        }

@tool("api_post", "Send data to API", {"endpoint": str, "data": dict})
async def api_post(args):
    try:
        result = await api.post(args["endpoint"], args["data"])
        return {"content": [{"type": "text", "text": str(result)}]}
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"API Error: {e}"}],
            "is_error": True
        }

# Create server
api_server = create_sdk_mcp_server(
    name="api_client",
    version="1.0.0",
    tools=[api_get, api_post]
)

{
    "content": [
        {"type": "text", "text": "Result text"},
        # Optional: image content
        {"type": "image", "data": "base64...", "mimeType": "image/png"}
    ],
    "is_error": False  # Optional: True if error
}

return {
    "content": [{"type": "text", "text": "Error message"}],
    "is_error": True
}