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 Guide

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

Create custom tools as Python functions that Claude can invoke using the MCP (Model Context Protocol) system.

Why Custom Tools?

Custom tools allow you to:

Extend Claude's capabilities with domain-specific operations
Integrate with external APIs and services
Add business logic validation
Control execution environment

Creating Your First Tool

Step 1: Define the Tool

from claude_agent_sdk import tool

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

Step 2: Create MCP Server

from claude_agent_sdk import create_sdk_mcp_server

server = create_sdk_mcp_server(
    name="greeter",
    version="1.0.0",
    tools=[greet]
)

Step 3: Use with Claude

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

options = ClaudeAgentOptions(
    mcp_servers={"greeter": server},
    allowed_tools=["mcp__greeter__greet"]  # Note the format!
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("Greet Alice")
    async for msg in client.receive_response():
        print(msg)

Important: MCP tools use the format mcp__<server_name>__<tool_name>

Tool Definition Patterns

Simple Tool

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

Tool with Complex 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...
    results = f"Found items for '{query}' (limit: {limit})"
    
    return {"content": [{"type": "text", "text": results}]}

Complete Example: Calculator Server

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

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

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

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

@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"{args['a'] / args['b']}"}]}

async def main():
    # Create server with all tools
    calculator = create_sdk_mcp_server(
        name="calculator",
        version="1.0.0",
        tools=[add, subtract, multiply, divide]
    )
    
    # Configure client
    options = ClaudeAgentOptions(
        mcp_servers={"calculator": calculator},
        allowed_tools=[
            "mcp__calculator__add",
            "mcp__calculator__subtract",
            "mcp__calculator__multiply",
            "mcp__calculator__divide"
        ]
    )
    
    async with ClaudeSDKClient(options=options) as client:
        await client.query("Calculate (10 + 5) * 3 - 8")
        async for msg in client.receive_response():
            print(msg)

anyio.run(main)

Tools with State

Create tools that access application state:

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",
    tools=[add_item, list_items, clear_items]
)

Tools with External APIs

import httpx

@tool("get_weather", "Get weather for a city", {"city": str})
async def get_weather(args):
    city = args["city"]
    
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.weather.example.com/current",
            params={"city": city, "api_key": "YOUR_KEY"}
        )
        
        if response.status_code == 200:
            data = response.json()
            return {
                "content": [{
                    "type": "text",
                    "text": f"Weather in {city}: {data['description']}, {data['temp']}°F"
                }]
            }
        else:
            return {
                "content": [{"type": "text", "text": f"Error: Could not fetch weather for {city}"}],
                "is_error": True
            }

Tools with Database Access

import asyncpg

async def get_db_connection():
    return await asyncpg.connect("postgresql://user:pass@localhost/db")

@tool("query_users", "Query users from database", {"name_filter": str})
async def query_users(args):
    conn = await get_db_connection()
    try:
        rows = await conn.fetch(
            "SELECT name, email FROM users WHERE name LIKE $1",
            f"%{args['name_filter']}%"
        )
        
        users = [f"{row['name']} ({row['email']})" for row in rows]
        result = "\n".join(users) if users else "No users found"
        
        return {"content": [{"type": "text", "text": result}]}
    finally:
        await conn.close()

Multiple MCP Servers

Combine multiple servers:

# Calculator server
calc_server = create_sdk_mcp_server("calc", tools=[add, subtract])

# Weather server
weather_server = create_sdk_mcp_server("weather", tools=[get_weather])

# Database server
db_server = create_sdk_mcp_server("database", tools=[query_users])

options = ClaudeAgentOptions(
    mcp_servers={
        "calc": calc_server,
        "weather": weather_server,
        "database": db_server
    },
    allowed_tools=[
        "mcp__calc__add",
        "mcp__calc__subtract",
        "mcp__weather__get_weather",
        "mcp__database__query_users"
    ]
)

Mixing SDK and External Servers

from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions

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

sdk_server = create_sdk_mcp_server("custom", tools=[custom_tool])

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

# Remote SSE server
remote_config = {
    "type": "sse",
    "url": "https://example.com/mcp",
    "headers": {"Authorization": "Bearer token"}
}

options = ClaudeAgentOptions(
    mcp_servers={
        "custom": sdk_server,
        "external": external_config,
        "remote": remote_config
    },
    allowed_tools=[
        "mcp__custom__custom_tool",
        "mcp__external__some_tool",
        "mcp__remote__remote_tool"
    ]
)

Best Practices

1. Clear Descriptions

# Good
@tool(
    "send_email",
    "Send an email to a recipient with subject and body",
    {"to": str, "subject": str, "body": str}
)

# Bad
@tool("email", "Sends email", {"to": str, "s": str, "b": str})

2. Validate Inputs

@tool("process_age", "Process user age", {"age": int})
async def process_age(args):
    age = args["age"]
    if age < 0 or age > 150:
        return {
            "content": [{"type": "text", "text": "Invalid age"}],
            "is_error": True
        }
    # Process valid age...

3. Handle Errors Gracefully

@tool("api_call", "Make API call", {"endpoint": str})
async def api_call(args):
    try:
        # Make API call...
        return {"content": [{"type": "text", "text": "Success"}]}
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Error: {str(e)}"}],
            "is_error": True
        }

4. Use Async for I/O

# Good (async for I/O operations)
@tool("fetch_data", "Fetch data", {"url": str})
async def fetch_data(args):
    async with httpx.AsyncClient() as client:
        response = await client.get(args["url"])
        return {"content": [{"type": "text", "text": response.text}]}

# Avoid (blocking I/O)
@tool("fetch_data_bad", "Fetch data", {"url": str})
async def fetch_data_bad(args):
    import requests
    response = requests.get(args["url"])  # Blocking!
    return {"content": [{"type": "text", "text": response.text}]}

Tool Response Format

Tools must return a dictionary with:

content: List of content blocks (required)
is_error: Boolean indicating error (optional)

# Success response
{
    "content": [
        {"type": "text", "text": "Result text"}
    ]
}

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

# Multiple content blocks
{
    "content": [
        {"type": "text", "text": "First part"},
        {"type": "text", "text": "Second part"}
    ]
}

Next Steps

Permission control: Permission Control Guide
Real examples: Real-World Scenarios
MCP reference: MCP Servers Reference

Install with Tessl CLI

npx tessl i tessl/pypi-claude-agent-sdk

docs

examples

guides

custom-tools.md

interactive-sessions.md

permission-control.md

quick-start.md

reference

index.md

tile.json