tessl/pypi-claude-agent-sdk

Python SDK for Claude Code enabling developers to build AI-powered applications and agents with support for custom tools, hooks, and bidirectional interactive conversations

Moderation error

Malicious code detected in tile.json: This tile describes a package named 'claude-agent-sdk' on PyPI, which appears to be typosquatting/impersonating official Anthropic SDK packages. The legitimate Anthropic Python SDK is 'anthropic', not 'claude-agent-sdk'. This naming pattern (claude-agent-sdk) is designed to deceive developers into installing a potentially malicious package by mimicking official Anthropic/Claude branding. This is a classic supply chain attack vector through package name confusion.

Overview

Eval results

Files

Custom Tools and MCP Servers

Name: tessl/pypi-claude-agent-sdk
Author: tessl

Create custom tools as Python functions that Claude can invoke, implemented as in-process MCP (Model Context Protocol) servers. SDK MCP servers run directly in your application process, providing better performance and simpler deployment compared to external MCP servers.

MCP Tool Naming: When using tools from MCP servers with allowed_tools or disallowed_tools, reference them with the format mcp__<server_name>__<tool_name>. For example, if you have a server named "calculator" with a tool "add", reference it as "mcp__calculator__add".

Capabilities

Tool Decorator

Define custom tools with type safety using the @tool decorator.

def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any]
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]

Parameters:

name: Unique identifier for the tool. Claude uses this to reference the tool
description: Human-readable description helping Claude understand when to use the tool
input_schema: Schema defining tool input parameters. Can be:
- Dictionary mapping parameter names to types (e.g., {"text": str})
- TypedDict class for complex schemas
- JSON Schema dictionary for full validation

Returns: Decorator function that wraps the tool implementation and returns SdkMcpTool instance

Usage:

Basic tool with simple schema:

from claude_agent_sdk import tool

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

Tool with multiple parameters:

@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 with error handling:

@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
        }
    return {
        "content": [{"type": "text", "text": f"Result: {args['a'] / args['b']}"}]
    }

Complex schema with JSON Schema:

@tool(
    "search",
    "Search for items",
    {
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "limit": {"type": "integer", "minimum": 1, "maximum": 100},
            "filters": {
                "type": "object",
                "properties": {
                    "category": {"type": "string"},
                    "min_price": {"type": "number"}
                }
            }
        },
        "required": ["query"]
    }
)
async def search(args):
    query = args["query"]
    limit = args.get("limit", 10)
    filters = args.get("filters", {})

    # Perform search logic
    results = f"Found items for '{query}' (limit: {limit}, filters: {filters})"

    return {
        "content": [{"type": "text", "text": results}]
    }

Tool Definition Class

The SdkMcpTool class represents a tool definition returned by the @tool decorator.

@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[[T], Awaitable[dict[str, Any]]]

Fields:

name: Tool identifier
description: Tool description
input_schema: Input parameter schema
handler: Async function implementing the tool

SDK MCP Server Factory

Create an in-process MCP server that runs within your Python application.

def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None
) -> McpSdkServerConfig

Parameters:

name: Unique identifier for the server
version: Server version string (default "1.0.0")
tools: List of SdkMcpTool instances created with @tool decorator

Returns: McpSdkServerConfig configuration object for use with ClaudeAgentOptions.mcp_servers

Usage:

Simple calculator server:

from claude_agent_sdk import tool, create_sdk_mcp_server

@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']}"}]}

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

Server with application state access:

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

store = DataStore()

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

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

@tool("clear_items", "Clear all items", {})
async def clear_items(args):
    count = len(store.items)
    store.items.clear()
    return {"content": [{"type": "text", "text": f"Cleared {count} items"}]}

server = create_sdk_mcp_server(
    name="store",
    version="1.0.0",
    tools=[add_item, list_items, clear_items]
)

Using SDK MCP Servers with Claude

Use SDK MCP servers with the client:

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server
)

@tool("weather", "Get weather for a city", {"city": str})
async def get_weather(args):
    city = args["city"]
    # In real implementation, fetch actual weather data
    return {
        "content": [{"type": "text", "text": f"Weather in {city}: Sunny, 72°F"}]
    }

@tool("time", "Get current time for a city", {"city": str})
async def get_time(args):
    city = args["city"]
    # In real implementation, fetch actual time
    return {
        "content": [{"type": "text", "text": f"Time in {city}: 3:45 PM"}]
    }

async def main():
    # Create MCP server
    server = create_sdk_mcp_server(
        name="info",
        version="1.0.0",
        tools=[get_weather, get_time]
    )

    # Configure client with server
    options = ClaudeAgentOptions(
        mcp_servers={"info": server},
        allowed_tools=["mcp__info__weather", "mcp__info__time"]
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("What's the weather in Paris?")
        async for msg in client.receive_response():
            print(msg)

        await client.query("What time is it in Tokyo?")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

MCP Server Configuration Types

SDK Server Config

Configuration for in-process SDK MCP servers.

class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: McpServer

Fields:

type: Always "sdk" for SDK servers
name: Server name
instance: MCP server instance (created by create_sdk_mcp_server)

External Server Configs

The SDK also supports external MCP servers running as separate processes:

Stdio server (subprocess):

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

SSE server (server-sent events):

class McpSSEServerConfig(TypedDict):
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]

HTTP server:

class McpHttpServerConfig(TypedDict):
    type: Literal["http"]
    url: str
    headers: NotRequired[dict[str, str]]

Union type for all server configs:

McpServerConfig = (
    McpStdioServerConfig |
    McpSSEServerConfig |
    McpHttpServerConfig |
    McpSdkServerConfig
)

Mixed Server Configurations

Use both SDK and external MCP servers together:

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    tool,
    create_sdk_mcp_server
)

# In-process SDK server
@tool("local_tool", "A local tool", {"input": str})
async def local_tool(args):
    return {"content": [{"type": "text", "text": f"Processed: {args['input']}"}]}

sdk_server = create_sdk_mcp_server("local", tools=[local_tool])

# Configure with both SDK and external servers
options = ClaudeAgentOptions(
    mcp_servers={
        "local": sdk_server,  # In-process SDK server
        "external": {  # External subprocess server
            "type": "stdio",
            "command": "external-mcp-server",
            "args": ["--config", "config.json"]
        },
        "remote": {  # Remote SSE server
            "type": "sse",
            "url": "https://example.com/mcp",
            "headers": {"Authorization": "Bearer token"}
        }
    },
    allowed_tools=[
        "mcp__local__local_tool",
        "mcp__external__some_tool",
        "mcp__remote__remote_tool"
    ]
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("Use the tools")
    async for msg in client.receive_response():
        print(msg)

Benefits of SDK MCP Servers

No subprocess management: Runs in the same process as your application, eliminating subprocess overhead and complexity

Better performance: No IPC (inter-process communication) overhead for tool calls, resulting in faster execution

Simpler deployment: Single Python process instead of managing multiple server processes

Easier debugging: All code runs in the same process, making debugging straightforward with standard Python debuggers

Type safety: Direct Python function calls with type hints and IDE support

State access: Tools have direct access to your application's variables and state without serialization

Migration from External Servers

Before (external MCP server):

options = ClaudeAgentOptions(
    mcp_servers={
        "calculator": {
            "type": "stdio",
            "command": "python",
            "args": ["-m", "calculator_server"]
        }
    }
)

After (SDK MCP server):

from my_tools import add, subtract  # Your tool functions

calculator = create_sdk_mcp_server(
    name="calculator",
    tools=[add, subtract]
)

options = ClaudeAgentOptions(
    mcp_servers={"calculator": calculator}
)

Complete Example

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

# Define tools
@tool("calculate_total", "Calculate total with tax", {
    "amount": float,
    "tax_rate": float
})
async def calculate_total(args):
    amount = args["amount"]
    tax_rate = args["tax_rate"]
    tax = amount * tax_rate
    total = amount + tax
    return {
        "content": [{
            "type": "text",
            "text": f"Amount: ${amount:.2f}, Tax: ${tax:.2f}, Total: ${total:.2f}"
        }]
    }

@tool("format_currency", "Format number as currency", {"amount": float})
async def format_currency(args):
    amount = args["amount"]
    return {
        "content": [{"type": "text", "text": f"${amount:,.2f}"}]
    }

async def main():
    # Create server
    server = create_sdk_mcp_server(
        name="finance",
        version="1.0.0",
        tools=[calculate_total, format_currency]
    )

    # Configure client
    options = ClaudeAgentOptions(
        mcp_servers={"finance": server},
        allowed_tools=["mcp__finance__calculate_total", "mcp__finance__format_currency"]
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query(
            "Calculate the total for a $100 purchase with 8.5% tax, "
            "then format the result as currency"
        )

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

anyio.run(main)

Install with Tessl CLI