Custom Tools

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

    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 either:
            - 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().
    """

@dataclass
class SdkMcpTool(Generic[T]):
    """Definition for an SDK MCP tool."""

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

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
        ClaudeCodeOptions.mcp_servers. This config contains the server
        instance and metadata needed for the SDK to route tool calls.
    """

from claude_code_sdk import tool, create_sdk_mcp_server, ClaudeCodeOptions, ClaudeSDKClient

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

async def main():
    # Create an SDK MCP server
    server = create_sdk_mcp_server(
        name="my-tools",
        version="1.0.0",
        tools=[greet_user]
    )

    # Use it with Claude
    options = ClaudeCodeOptions(
        mcp_servers={"tools": server},
        allowed_tools=["mcp__tools__greet"]  # Note the naming convention
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Greet Alice")

        async for msg in client.receive_response():
            print(msg)

@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("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply_numbers(args):
    result = args["a"] * args["b"]
    return {
        "content": [
            {"type": "text", "text": f"Product: {result}"}
        ]
    }

async def main():
    calculator = create_sdk_mcp_server(
        name="calculator",
        version="2.0.0",
        tools=[add_numbers, multiply_numbers]
    )

    options = ClaudeCodeOptions(
        mcp_servers={"calc": calculator},
        allowed_tools=["mcp__calc__add", "mcp__calc__multiply"]
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Calculate 15 + 27, then multiply the result by 3")

        async for msg in client.receive_response():
            print(msg)

@tool("divide", "Divide two numbers", {"a": float, "b": float})
async def divide_numbers(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_extensions import TypedDict

class SearchParameters(TypedDict):
    query: str
    max_results: int
    include_metadata: bool

@tool("search", "Search for items", SearchParameters)
async def search_items(args):
    # Access typed parameters
    query = args["query"]
    max_results = args.get("max_results", 10)
    include_metadata = args.get("include_metadata", False)

    # Perform search logic
    results = perform_search(query, max_results, include_metadata)

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

# Alternative: JSON Schema approach
json_schema = {
    "type": "object",
    "properties": {
        "query": {"type": "string", "description": "Search query"},
        "max_results": {"type": "integer", "minimum": 1, "maximum": 100, "default": 10},
        "include_metadata": {"type": "boolean", "default": False}
    },
    "required": ["query"]
}

@tool("advanced_search", "Advanced search with JSON schema", json_schema)
async def advanced_search(args):
    # Implementation
    pass

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

# Global application state
app_store = DataStore()

@tool("add_item", "Add item to store", {"item": str})
async def add_item(args):
    app_store.items.append(args["item"])
    app_store.counter += 1
    return {
        "content": [
            {"type": "text", "text": f"Added: {args['item']} (total: {app_store.counter})"}
        ]
    }

@tool("list_items", "List all items in store", {})
async def list_items(args):
    if not app_store.items:
        return {
            "content": [
                {"type": "text", "text": "No items in store"}
            ]
        }

    items_text = "\n".join(f"- {item}" for item in app_store.items)
    return {
        "content": [
            {"type": "text", "text": f"Items in store:\n{items_text}"}
        ]
    }

async def main():
    server = create_sdk_mcp_server(
        name="store",
        tools=[add_item, list_items]
    )

    options = ClaudeCodeOptions(
        mcp_servers={"store": server},
        allowed_tools=["mcp__store__add_item", "mcp__store__list_items"]
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Add 'apple' to the store, then list all items")

        async for msg in client.receive_response():
            print(msg)

# SDK server for in-process tools
internal_server = create_sdk_mcp_server(
    name="internal",
    tools=[my_custom_tool]
)

# External server configuration
external_server_config = {
    "type": "stdio",
    "command": "external-mcp-server",
    "args": ["--config", "config.json"]
}

options = ClaudeCodeOptions(
    mcp_servers={
        "internal": internal_server,      # In-process SDK server
        "external": external_server_config  # External subprocess server
    },
    allowed_tools=[
        "mcp__internal__my_custom_tool",
        "mcp__external__some_external_tool"
    ]
)

{
    "content": [
        {"type": "text", "text": "Response text"}
    ],
    "is_error": False  # Optional, defaults to False
}