What this does

Custom MCP tools expose your application's APIs, database queries, or business logic as discoverable tools that any MCP-compatible LLM client can invoke. Each tool is an async function with typed parameters.

Steps

Define a tool with complex parameters. Use typed parameters with defaults and optional fields.

from mcp.server import Server, stdio_server
from typing import Optional

server = Server("custom-app-tools")

@server.tool("search_products")
async def search_products(
    query: str,
    category: Optional[str] = None,
    max_price: Optional[float] = None,
    sort_by: str = "relevance",
    limit: int = 10
) -> str:
    """Search products in the inventory database."""
    # Application logic here
    results = [{"id": 1, "name": "Widget", "price": 9.99}]
    return str(results)

Inject application dependencies. Use a factory pattern to create tools with access to shared resources.

from dataclasses import dataclass

@dataclass
class AppContext:
    db_connection: str
    api_key: str

def create_tools(ctx: AppContext):
    server = Server("app-tools")

    @server.tool("query_database")
    async def query_database(sql: str) -> str:
        # ctx.db_connection is available
        return f"Executed: {sql} on {ctx.db_connection}"

    return server

Return structured data as JSON strings. MCP tool responses are text; serialize complex data.

import json

@server.tool("get_user_report")
async def get_user_report(user_id: int) -> str:
    report = {
        "user_id": user_id,
        "orders": [{"id": 101, "total": 49.99}],
        "metrics": {"visit_count": 42}
    }
    return json.dumps(report, indent=2)

Add validation using Pydantic models.

from pydantic import BaseModel, Field

class OrderInput(BaseModel):
    user_id: int = Field(gt=0)
    items: list[str] = Field(min_length=1)
    coupon: Optional[str] = None

@server.tool("place_order")
async def place_order(user_id: int, items: list[str], coupon: str = None) -> str:
    validated = OrderInput(user_id=user_id, items=items, coupon=coupon)
    return f"Order placed for user {validated.user_id}"

Handle tool-specific errors. Return meaningful error messages.

@server.tool("delete_user")
async def delete_user(user_id: int) -> str:
    if user_id <= 0:
        return json.dumps({"error": "Invalid user ID"})
    try:
        # Perform deletion
        return json.dumps({"success": True, "deleted_id": user_id})
    except Exception as e:
        return json.dumps({"error": str(e)})

Verification

python -c "
from mcp.server import Server
s = Server('custom')

@s.tool('test_tool')
async def test_tool(x: int) -> str:
    return str(x * 2)

print(list(s._tool_registry.keys()))
# Expected: ['test_tool']
"

Common failures

Tool parameters must be JSON-serializable. Python types like datetime or set cause serialization errors. Convert to strings before returning.
Async function not awaited. The SDK expects async functions. Defining def instead of async def causes registration to fail silently.
Tool name conflicts. Two tools with the same name overwrite each other. Use a prefix like app_search_products to namespace.
Version mismatch - The installed package or runtime differs from the command shown; check the version first and rerun the smallest verification command.
Local environment drift - Another service, virtual environment, model, or path is being used; print the active binary path and configuration before changing the guide steps.

Related guides

How to Set Up MCP Server with Standard Tools
How to Connect MCP Client to Remote MCP Server