Mcp Server Building

MCP Server Building

Build custom MCP servers to extend Claude with tools, resources, and prompts.

Architecture

+-------------+     JSON-RPC      +-------------+
|   Claude    |<----------------->| MCP Server  |
|   (Host)    |   stdio/SSE/WS    |  (Tools)    |
+-------------+                   +-------------+

Three Primitives:

• Tools: Functions Claude can call (with user approval)
• Resources: Data Claude can read (files, API responses)
• Prompts: Pre-defined prompt templates

Quick Start

Minimal Python Server (stdio)

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("my-tools")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="greet",
            description="Greet a user by name",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "Name to greet"}
                },
                "required": ["name"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "greet":
        return [TextContent(type="text", text=f"Hello, {arguments['name']}!")]
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

TypeScript Server (production)

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
  { name: "my-tools", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [{
    name: "fetch_url",
    description: "Fetch content from a URL",
    inputSchema: {
      type: "object",
      properties: { url: { type: "string" } },
      required: ["url"],
    },
  }],
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  if (request.params.name === "fetch_url") {
    const { url } = request.params.arguments as { url: string };
    const response = await fetch(url);
    return { content: [{ type: "text", text: await response.text() }] };
  }
  throw new Error("Unknown tool");
});

await server.connect(new StdioServerTransport());

Detailed Guides

• Transport patterns: See references/transport-patterns.md for stdio, SSE, WebSocket
• Tool definitions: See references/tool-definitions.md for schemas, error handling, caching
• Resource patterns: See references/resource-patterns.md for files and dynamic data
• Testing: See references/testing-patterns.md for MCP Inspector and pytest
• Auto-discovery: See references/auto-discovery.md for CC 2.1.7+ optimization

Key Decisions

Anti-Patterns

1. Stateful tools without cleanup - Always clean up connections 2. Blocking synchronous code - Use asyncio.to_thread() 3. Missing input validation - Validate before processing 4. Secrets in tool output - Never return credentials 5. Unbounded responses - Limit response sizes

Related Skills

• function-calling - LLM function calling patterns
• agent-loops - Agentic patterns using MCP tools
• input-validation - Input validation for arguments

Resources

• MCP Specification: https://modelcontextprotocol.io/docs
• Python SDK: https://github.com/modelcontextprotocol/python-sdk
• TypeScript SDK: https://github.com/modelcontextprotocol/typescript-sdk

MCP Auto-Discovery Optimization

Claude Code 2.1.7+ uses automatic MCP discovery via MCPSearch. When context exceeds 10%, your MCP tools are discovered on-demand rather than pre-loaded.

Optimizing for Auto-Discovery

Use descriptive names and keywords:

# GOOD: Descriptive, searchable
Tool(
    name="query_product_database",
    description="""
    Search the product catalog database.

    KEYWORDS: products, catalog, inventory, SKU, search
    USE WHEN: User needs product info, pricing, availability
    """,
    inputSchema={...}
)

# BAD: Generic, hard to discover
Tool(
    name="search",
    description="Search things",
    inputSchema={...}
)

Token-Efficient Tool Definitions

Tool definitions consume context when loaded. Optimize for size:

# Verbose: ~200 tokens
Tool(
    name="search_database",
    description="This tool allows you to search our comprehensive database...",
    inputSchema={...}  # detailed descriptions
)

# Concise: ~80 tokens
Tool(
    name="search_database",
    description="Search database. Supports: full-text, filters. Returns: {id, title, snippet}",
    inputSchema={...}  # brief descriptions
)

Discovery Metadata Pattern

Add discovery hints to improve MCPSearch matching:

Tool(
    name="analyze_logs",
    description="""
    Analyze application logs for errors.

    Category: Observability
    Keywords: logs, errors, debugging, monitoring
    Triggers: "check logs", "find errors", "debug issue"
    """,
    inputSchema={...}
)

Best Practices for Discovery

MCP Resource Patterns

Resources are data that Claude can read (files, API responses, etc).

File Resources

@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="file:///config/settings.json",
            name="Settings",
            mimeType="application/json",
            description="Application configuration"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "file:///config/settings.json":
        return Path("settings.json").read_text()
    raise ValueError(f"Unknown resource: {uri}")

Dynamic Resources (API Data)

@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="api://users/current",
            name="Current User",
            mimeType="application/json"
        ),
        Resource(
            uri="api://metrics/today",
            name="Today's Metrics",
            mimeType="application/json"
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri.startswith("api://"):
        endpoint = uri.replace("api://", "")
        data = await api_client.get(endpoint)
        return json.dumps(data, indent=2)

Resource vs Tool Decision

Resource URI Schemes

• file:// - Local filesystem
• api:// - API endpoints
• db:// - Database queries
• mem:// - In-memory data

MCP Testing Patterns

Manual Testing with Inspector

# Test with MCP Inspector
npx @modelcontextprotocol/inspector python server.py

The Inspector provides an interactive UI to:

• List available tools and resources
• Call tools with custom arguments
• View responses and errors

Automated Testing

import pytest
from mcp.client import Client
from mcp.client.stdio import stdio_client

@pytest.mark.asyncio
async def test_greet_tool():
    async with stdio_client("python", ["server.py"]) as (read, write):
        client = Client("test", "1.0.0")
        await client.connect(read, write)

        # List tools
        tools = await client.list_tools()
        assert any(t.name == "greet" for t in tools.tools)

        # Call tool
        result = await client.call_tool("greet", {"name": "World"})
        assert "Hello, World!" in result.content[0].text

Integration Testing Pattern

@pytest.fixture
async def mcp_client():
    """Create MCP client for testing."""
    async with stdio_client("python", ["server.py"]) as (read, write):
        client = Client("test", "1.0.0")
        await client.connect(read, write)
        yield client

@pytest.mark.asyncio
async def test_search_returns_results(mcp_client):
    result = await mcp_client.call_tool("search", {"query": "test"})
    data = json.loads(result.content[0].text)
    assert len(data["results"]) > 0

@pytest.mark.asyncio
async def test_search_handles_empty_query(mcp_client):
    result = await mcp_client.call_tool("search", {"query": ""})
    assert "Error" in result.content[0].text

Testing Error Conditions

@pytest.mark.asyncio
async def test_handles_api_timeout(mcp_client, mock_api):
    mock_api.delay = 30  # Simulate timeout
    result = await mcp_client.call_tool("fetch_data", {"id": "123"})
    assert "timeout" in result.content[0].text.lower()

@pytest.mark.asyncio
async def test_handles_invalid_input(mcp_client):
    result = await mcp_client.call_tool("process", {"data": None})
    assert "Error" in result.content[0].text

MCP Tool Definition Patterns

Input Schema Best Practices

Tool(
    name="search_database",
    description="Search the product database. Returns up to 10 results.",
    inputSchema={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Search query (supports wildcards with *)"
            },
            "category": {
                "type": "string",
                "enum": ["electronics", "clothing", "books"],
                "description": "Filter by category"
            },
            "max_results": {
                "type": "integer",
                "minimum": 1,
                "maximum": 50,
                "default": 10,
                "description": "Maximum results to return"
            }
        },
        "required": ["query"]
    }
)

Guidelines:

• Always include description for each property
• Use enum for fixed option sets
• Set minimum/maximum for numbers
• Mark required fields explicitly
• Provide default values where sensible

Error Handling

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    try:
        if name == "query_api":
            result = await external_api.query(arguments["query"])
            return [TextContent(type="text", text=json.dumps(result))]
    except ExternalAPIError as e:
        # Return error as text - Claude will see and handle it
        return [TextContent(
            type="text",
            text=f"Error: API returned {e.status_code}: {e.message}"
        )]
    except Exception as e:
        # Log internally, return user-friendly message
        logger.exception("Tool execution failed")
        return [TextContent(
            type="text",
            text=f"Error: {type(e).__name__}: {str(e)}"
        )]

Caching Expensive Operations

from datetime import datetime, timedelta

_cache = {}
_cache_ttl = timedelta(minutes=5)

async def get_cached_data(key: str) -> dict:
    now = datetime.now()
    if key in _cache:
        data, timestamp = _cache[key]
        if now - timestamp < _cache_ttl:
            return data

    data = await expensive_fetch(key)
    _cache[key] = (data, now)
    return data

Rate Limiting

import asyncio
from collections import defaultdict

_request_times = defaultdict(list)
MAX_REQUESTS_PER_MINUTE = 60

async def rate_limited_call(user_id: str, func, *args):
    now = asyncio.get_event_loop().time()
    _request_times[user_id] = [
        t for t in _request_times[user_id]
        if now - t < 60
    ]

    if len(_request_times[user_id]) >= MAX_REQUESTS_PER_MINUTE:
        raise Exception("Rate limit exceeded. Try again in a minute.")

    _request_times[user_id].append(now)
    return await func(*args)

Anti-Patterns

1. Stateful tools without cleanup: Always clean up connections/resources 2. Blocking synchronous code: Use asyncio.to_thread() for blocking ops 3. Missing input validation: Always validate before processing 4. Secrets in tool output: Never return API keys or credentials 5. Unbounded responses: Limit response sizes (Claude has context limits)

MCP Transport Patterns

stdio Transport (CLI)

Standard I/O is the simplest transport for CLI tools and Claude Desktop.

Python stdio Server

from mcp.server import Server
from mcp.server.stdio import stdio_server

server = Server("my-tools")

async def main():
    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

TypeScript stdio Server

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const server = new Server(
  { name: "my-tools", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

const transport = new StdioServerTransport();
await server.connect(transport);

SSE Transport (Web)

Server-Sent Events for browser and web deployments.

from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route

sse = SseServerTransport("/messages")

async def handle_sse(request):
    async with sse.connect_sse(
        request.scope, request.receive, request._send
    ) as streams:
        await server.run(
            streams[0], streams[1],
            server.create_initialization_options()
        )

app = Starlette(routes=[
    Route("/sse", endpoint=handle_sse),
    Route("/messages", endpoint=sse.handle_post_message, methods=["POST"]),
])

Claude Desktop Configuration

// macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
// Windows: %APPDATA%\Claude\claude_desktop_config.json
{
  "mcpServers": {
    "database": {
      "command": "npx",
      "args": ["-y", "@myorg/db-tools"],
      "env": {
        "DATABASE_URL": "postgres://..."
      }
    },
    "python-tools": {
      "command": "uv",
      "args": ["run", "python", "-m", "my_mcp_server"],
      "cwd": "/path/to/project"
    }
  }
}

Transport Decision Matrix