0
# Custom Tools (In-Process MCP Servers)
1

2
Create custom tools as Python functions that Claude can invoke, running in-process without subprocess overhead. The SDK provides decorators and helpers for building Model Context Protocol (MCP) servers directly in your Python application.
3

4
## Capabilities
5

6
### Tool Decorator
7

8
Define custom tools using the `@tool` decorator.
9

10
```python { .api }
11
def tool(
12
    name: str,
13
    description: str,
14
    input_schema: type | dict[str, Any]
15
) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]:
16
    """
17
    Decorator for creating custom tools.
18

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

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

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

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

45
**Parameters:**
46

47
- `name` (str): Unique identifier for the tool. Claude uses this name to reference the tool in function calls.
48

49
- `description` (str): Human-readable description helping Claude understand when and how to use the tool.
50

51
- `input_schema` (type | dict[str, Any]): Schema defining input parameters. Options:
52
  - Simple dict: `{"param_name": type}` (e.g., `{"text": str, "count": int}`)
53
  - TypedDict class for structured schemas
54
  - Full JSON Schema dict with validation rules
55

56
**Returns:**
57

58
Decorator that wraps the async handler function and returns an `SdkMcpTool` instance.
59

60
**Usage Example - Basic Tool:**
61

62
```python
63
from claude_agent_sdk import tool
64

65
@tool("greet", "Greet a user by name", {"name": str})
66
async def greet(args):
67
    return {
68
        "content": [
69
            {"type": "text", "text": f"Hello, {args['name']}!"}
70
        ]
71
    }
72
```
73

74
**Usage Example - Multiple Parameters:**
75

76
```python
77
@tool("add", "Add two numbers", {"a": float, "b": float})
78
async def add_numbers(args):
79
    result = args["a"] + args["b"]
80
    return {
81
        "content": [
82
            {"type": "text", "text": f"Result: {result}"}
83
        ]
84
    }
85
```
86

87
**Usage Example - Error Handling:**
88

89
```python
90
@tool("divide", "Divide two numbers", {"a": float, "b": float})
91
async def divide(args):
92
    if args["b"] == 0:
93
        return {
94
            "content": [{"type": "text", "text": "Error: Division by zero"}],
95
            "is_error": True
96
        }
97
    result = args["a"] / args["b"]
98
    return {
99
        "content": [{"type": "text", "text": f"Result: {result}"}]
100
    }
101
```
102

103
**Usage Example - Complex Schema:**
104

105
```python
106
from typing import TypedDict
107

108
class SearchInput(TypedDict):
109
    query: str
110
    limit: int
111
    filters: dict[str, str]
112

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

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

122
    return {
123
        "content": [
124
            {"type": "text", "text": f"Found {len(results)} results"}
125
        ]
126
    }
127
```
128

129
**Usage Example - JSON Schema:**
130

131
```python
132
schema = {
133
    "type": "object",
134
    "properties": {
135
        "text": {"type": "string", "minLength": 1},
136
        "count": {"type": "integer", "minimum": 0, "maximum": 100}
137
    },
138
    "required": ["text"]
139
}
140

141
@tool("repeat", "Repeat text multiple times", schema)
142
async def repeat_text(args):
143
    text = args["text"]
144
    count = args.get("count", 1)
145
    return {
146
        "content": [
147
            {"type": "text", "text": text * count}
148
        ]
149
    }
150
```
151

152
### Create MCP Server
153

154
Create an in-process MCP server with custom tools.
155

156
```python { .api }
157
def create_sdk_mcp_server(
158
    name: str,
159
    version: str = "1.0.0",
160
    tools: list[SdkMcpTool[Any]] | None = None
161
) -> McpSdkServerConfig:
162
    """
163
    Create an in-process MCP server that runs within your Python application.
164

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

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

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

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

194
**Parameters:**
195

196
- `name` (str): Unique identifier for the server used in `mcp_servers` configuration.
197

198
- `version` (str): Server version string for informational purposes. Default: "1.0.0".
199

200
- `tools` (list[SdkMcpTool[Any]] | None): List of tool instances created with `@tool` decorator.
201

202
**Returns:**
203

204
`McpSdkServerConfig` for use in `ClaudeAgentOptions.mcp_servers`.
205

206
**Usage Example - Simple Server:**
207

208
```python
209
from claude_agent_sdk import tool, create_sdk_mcp_server, ClaudeAgentOptions
210

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

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

219
# Create server
220
calculator = create_sdk_mcp_server(
221
    name="calculator",
222
    version="2.0.0",
223
    tools=[add, multiply]
224
)
225

226
# Use with Claude
227
options = ClaudeAgentOptions(
228
    mcp_servers={"calc": calculator},
229
    allowed_tools=["add", "multiply"]
230
)
231
```
232

233
**Usage Example - Server with Application State:**
234

235
```python
236
from claude_agent_sdk import tool, create_sdk_mcp_server
237

238
class DataStore:
239
    def __init__(self):
240
        self.items = []
241

242
    def add_item(self, item):
243
        self.items.append(item)
244

245
    def get_items(self):
246
        return self.items
247

248
# Create store instance
249
store = DataStore()
250

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

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

262
# Create server
263
server = create_sdk_mcp_server("store", tools=[add_item, list_items])
264
```
265

266
### SdkMcpTool
267

268
Tool definition dataclass (typically created by `@tool` decorator).
269

270
```python { .api }
271
@dataclass
272
class SdkMcpTool(Generic[T]):
273
    """Custom tool definition."""
274

275
    name: str
276
    description: str
277
    input_schema: type[T] | dict[str, Any]
278
    handler: Callable[[T], Awaitable[dict[str, Any]]]
279
```
280

281
**Fields:**
282

283
- `name` (str): Tool name.
284

285
- `description` (str): Tool description for Claude.
286

287
- `input_schema` (type[T] | dict[str, Any]): Input schema as type or dict.
288

289
- `handler` (Callable[[T], Awaitable[dict[str, Any]]]): Async handler function.
290

291
**Note:** Typically you create tools using the `@tool` decorator rather than instantiating `SdkMcpTool` directly.
292

293
**Manual Usage Example:**
294

295
```python
296
from claude_agent_sdk import SdkMcpTool
297

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

301
tool_instance = SdkMcpTool(
302
    name="my_tool",
303
    description="Does something",
304
    input_schema={"param": str},
305
    handler=my_handler
306
)
307
```
308

309
## Complete Examples
310

311
### Calculator Server
312

313
```python
314
from claude_agent_sdk import (
315
    tool, create_sdk_mcp_server, query,
316
    ClaudeAgentOptions, AssistantMessage, TextBlock
317
)
318
import anyio
319

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

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

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

335
@tool("divide", "Divide two numbers", {"a": float, "b": float})
336
async def divide(args):
337
    if args["b"] == 0:
338
        return {
339
            "content": [{"type": "text", "text": "Error: Division by zero"}],
340
            "is_error": True
341
        }
342
    result = args["a"] / args["b"]
343
    return {"content": [{"type": "text", "text": f"{result}"}]}
344

345
# Create calculator server
346
calc_server = create_sdk_mcp_server(
347
    name="calculator",
348
    version="1.0.0",
349
    tools=[add, subtract, multiply, divide]
350
)
351

352
# Use calculator
353
async def main():
354
    options = ClaudeAgentOptions(
355
        mcp_servers={"calc": calc_server},
356
        allowed_tools=["add", "subtract", "multiply", "divide"]
357
    )
358

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

365
anyio.run(main)
366
```
367

368
### Database Server
369

370
```python
371
from claude_agent_sdk import tool, create_sdk_mcp_server
372
from typing import TypedDict, Literal
373

374
class QueryInput(TypedDict):
375
    sql: str
376
    params: dict[str, Any]
377

378
class InsertInput(TypedDict):
379
    table: str
380
    data: dict[str, Any]
381

382
class Database:
383
    def __init__(self):
384
        self.tables = {}
385

386
    async def query(self, sql: str, params: dict):
387
        # Execute query
388
        pass
389

390
    async def insert(self, table: str, data: dict):
391
        # Insert data
392
        if table not in self.tables:
393
            self.tables[table] = []
394
        self.tables[table].append(data)
395

396
# Create database instance
397
db = Database()
398

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

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

409
# Create server
410
db_server = create_sdk_mcp_server(
411
    name="database",
412
    version="1.0.0",
413
    tools=[db_query, db_insert]
414
)
415
```
416

417
### File Processing Server
418

419
```python
420
from claude_agent_sdk import tool, create_sdk_mcp_server
421
import hashlib
422

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

428
    try:
429
        hasher = hashlib.new(algorithm)
430
        with open(path, "rb") as f:
431
            for chunk in iter(lambda: f.read(4096), b""):
432
                hasher.update(chunk)
433
        return {
434
            "content": [{"type": "text", "text": f"{algorithm}: {hasher.hexdigest()}"}]
435
        }
436
    except Exception as e:
437
        return {
438
            "content": [{"type": "text", "text": f"Error: {e}"}],
439
            "is_error": True
440
        }
441

442
@tool("file_stats", "Get file statistics", {"path": str})
443
async def file_stats(args):
444
    import os
445
    try:
446
        stats = os.stat(args["path"])
447
        return {
448
            "content": [{
449
                "type": "text",
450
                "text": f"Size: {stats.st_size} bytes, Modified: {stats.st_mtime}"
451
            }]
452
        }
453
    except Exception as e:
454
        return {
455
            "content": [{"type": "text", "text": f"Error: {e}"}],
456
            "is_error": True
457
        }
458

459
# Create server
460
file_server = create_sdk_mcp_server(
461
    name="file_tools",
462
    version="1.0.0",
463
    tools=[hash_file, file_stats]
464
)
465
```
466

467
### API Client Server
468

469
```python
470
from claude_agent_sdk import tool, create_sdk_mcp_server
471
import httpx
472

473
class APIClient:
474
    def __init__(self, base_url: str, api_key: str):
475
        self.base_url = base_url
476
        self.api_key = api_key
477

478
    async def get(self, endpoint: str):
479
        async with httpx.AsyncClient() as client:
480
            response = await client.get(
481
                f"{self.base_url}{endpoint}",
482
                headers={"Authorization": f"Bearer {self.api_key}"}
483
            )
484
            return response.json()
485

486
    async def post(self, endpoint: str, data: dict):
487
        async with httpx.AsyncClient() as client:
488
            response = await client.post(
489
                f"{self.base_url}{endpoint}",
490
                json=data,
491
                headers={"Authorization": f"Bearer {self.api_key}"}
492
            )
493
            return response.json()
494

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

498
@tool("api_get", "Fetch data from API", {"endpoint": str})
499
async def api_get(args):
500
    try:
501
        data = await api.get(args["endpoint"])
502
        return {"content": [{"type": "text", "text": str(data)}]}
503
    except Exception as e:
504
        return {
505
            "content": [{"type": "text", "text": f"API Error: {e}"}],
506
            "is_error": True
507
        }
508

509
@tool("api_post", "Send data to API", {"endpoint": str, "data": dict})
510
async def api_post(args):
511
    try:
512
        result = await api.post(args["endpoint"], args["data"])
513
        return {"content": [{"type": "text", "text": str(result)}]}
514
    except Exception as e:
515
        return {
516
            "content": [{"type": "text", "text": f"API Error: {e}"}],
517
            "is_error": True
518
        }
519

520
# Create server
521
api_server = create_sdk_mcp_server(
522
    name="api_client",
523
    version="1.0.0",
524
    tools=[api_get, api_post]
525
)
526
```
527

528
## Tool Response Format
529

530
Tool handlers must return a dict with specific structure:
531

532
```python
533
{
534
    "content": [
535
        {"type": "text", "text": "Result text"},
536
        # Optional: image content
537
        {"type": "image", "data": "base64...", "mimeType": "image/png"}
538
    ],
539
    "is_error": False  # Optional: True if error
540
}
541
```
542

543
**Content Types:**
544

545
- Text: `{"type": "text", "text": "..."}`
546
- Image: `{"type": "image", "data": "base64data", "mimeType": "image/png"}`
547

548
**Error Responses:**
549

550
Set `"is_error": True` to indicate failure:
551

552
```python
553
return {
554
    "content": [{"type": "text", "text": "Error message"}],
555
    "is_error": True
556
}
557
```
558

559
## Best Practices
560

561
1. **Descriptive Names**: Use clear, action-oriented tool names (e.g., "search_database", not "tool1")
562

563
2. **Detailed Descriptions**: Write descriptions that help Claude understand when to use the tool and what it does
564

565
3. **Type Safety**: Use TypedDict or full JSON schemas for complex inputs
566

567
4. **Error Handling**: Always handle exceptions and return error responses with `is_error: True`
568

569
5. **State Access**: Tools can access application state directly (closure over variables)
570

571
6. **Async/Await**: All tool handlers must be async functions
572

573
7. **Validation**: Validate inputs in handler, even with schema validation
574

575
8. **Documentation**: Document tool behavior in function docstrings
576

577
9. **Testing**: Test tools independently before using with Claude
578

579
10. **Performance**: In-process tools are fast; prefer them over subprocess MCP servers
580

581
## Advantages of In-Process MCP Servers
582

583
- **Performance**: No IPC overhead, direct function calls
584
- **Simplicity**: Single process deployment
585
- **Debugging**: Standard Python debugging works
586
- **State Access**: Direct access to application variables
587
- **Type Safety**: Full Python type hints and IDE support
588
- **Error Handling**: Python exception handling
589
- **Testing**: Standard Python testing frameworks
590