Building an MCP Server with the Python SDK

The Model Context Protocol (MCP) is the standard for connecting AI applications to external systems. Building an MCP server lets you expose your APIs, databases, and tools to any MCP-compatible AI agent — from Claude to ChatGPT to custom builds. This guide walks through building a complete MCP server with the Python SDK.

What MCP Enables

MCP gives AI applications a standardized way to:

• Discover tools — Agents learn what tools are available and how to call them
• Execute functions — Agents invoke your tools with structured inputs
• Read resources — Agents access your data through a file-like interface
• Use prompts — Agents leverage pre-built prompt templates

Think of it as building an API, but specifically designed for AI consumption.

Setting Up the Project

# Create project directory
uv init my-mcp-server
cd my-mcp-server

# Set up virtual environment
uv venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# Install MCP SDK
uv add "mcp[cli]"

# Create server file
touch server.py

Building Your First MCP Server

Here’s a complete MCP server that exposes a task management system:

from mcp.server import MCPServer
from typing import Optional
from datetime import datetime
import json

# Initialize the MCP server
mcp = MCPServer("task-manager")

# In-memory task store (use a database in production)
tasks: dict[str, dict] = {}
task_counter = 0

@mcp.tool()
async def create_task(
    title: str,
    description: str = "",
    priority: str = "medium",
    due_date: Optional[str] = None
) -> str:
    """Create a new task with title, description, and priority.
    
    Args:
        title: The task title
        description: Optional task description
        priority: Task priority (low, medium, high, critical)
        due_date: Optional due date in ISO format (YYYY-MM-DD)
    """
    global task_counter
    task_counter += 1
    
    task_id = f"task-{task_counter}"
    tasks[task_id] = {
        "id": task_id,
        "title": title,
        "description": description,
        "priority": priority,
        "status": "pending",
        "due_date": due_date,
        "created_at": datetime.now().isoformat(),
    }
    
    return json.dumps({
        "success": True,
        "task": tasks[task_id],
        "message": f"Task '{title}' created with ID {task_id}"
    })

@mcp.tool()
async def list_tasks(
    status: Optional[str] = None,
    priority: Optional[str] = None
) -> str:
    """List all tasks, optionally filtered by status or priority.
    
    Args:
        status: Filter by status (pending, in_progress, completed)
        priority: Filter by priority (low, medium, high, critical)
    """
    filtered = tasks.values()
    
    if status:
        filtered = [t for t in filtered if t["status"] == status]
    if priority:
        filtered = [t for t in filtered if t["priority"] == priority]
    
    return json.dumps({
        "tasks": list(filtered),
        "count": len(list(filtered))
    })

@mcp.tool()
async def update_task(
    task_id: str,
    title: Optional[str] = None,
    status: Optional[str] = None,
    priority: Optional[str] = None
) -> str:
    """Update an existing task's properties.
    
    Args:
        task_id: The ID of the task to update
        title: New title (optional)
        status: New status (optional)
        priority: New priority (optional)
    """
    if task_id not in tasks:
        return json.dumps({"error": f"Task {task_id} not found"})
    
    task = tasks[task_id]
    if title:
        task["title"] = title
    if status:
        task["status"] = status
    if priority:
        task["priority"] = priority
    
    return json.dumps({"success": True, "task": task})

@mcp.tool()
async def delete_task(task_id: str) -> str:
    """Delete a task by ID.
    
    Args:
        task_id: The ID of the task to delete
    """
    if task_id not in tasks:
        return json.dumps({"error": f"Task {task_id} not found"})
    
    deleted = tasks.pop(task_id)
    return json.dumps({
        "success": True,
        "message": f"Task '{deleted['title']}' deleted"
    })

Exposing Resources

Resources provide read-only access to your data:

@mcp.resource("tasks://summary")
async def get_task_summary() -> str:
    """Get a summary of all tasks by status and priority."""
    summary = {
        "total": len(tasks),
        "by_status": {},
        "by_priority": {}
    }
    
    for task in tasks.values():
        status = task["status"]
        priority = task["priority"]
        summary["by_status"][status] = summary["by_status"].get(status, 0) + 1
        summary["by_priority"][priority] = summary["by_priority"].get(priority, 0) + 1
    
    return json.dumps(summary, indent=2)

@mcp.resource("tasks://overdue")
async def get_overdue_tasks() -> str:
    """Get all tasks that are past their due date."""
    now = datetime.now()
    overdue = [
        task for task in tasks.values()
        if task["due_date"] and datetime.fromisoformat(task["due_date"]) < now
    ]
    return json.dumps(overdue, indent=2)

Adding Prompts

Prompts provide templates that help users accomplish specific tasks:

@mcp.prompt()
async def task_review_prompt(project_name: str) -> str:
    """Generate a prompt for reviewing project tasks.
    
    Args:
        project_name: The name of the project to review
    """
    return f"""Please review the following tasks for project '{project_name}':

1. List all pending tasks and their priorities
2. Identify any overdue tasks
3. Suggest task prioritization for the next sprint
4. Flag any tasks that may need more details

Provide a structured summary with actionable recommendations."""

@mcp.prompt()
async def daily_standup_prompt(team_member: str) -> str:
    """Generate a standup update prompt.
    
    Args:
        team_member: Name of the team member
    """
    return f"""Help me prepare a standup update for {team_member}:

1. What tasks were completed yesterday?
2. What tasks are planned for today?
3. Are there any blockers or dependencies?
4. What's the overall progress percentage?

Format as a concise standup update."""

Running and Testing

Start the Server

# Run with stdio transport (for Claude for Desktop)
uv run server.py

# Run with HTTP transport (for web clients)
mcp run server.py --transport http --port 8080

Configure Claude for Desktop

Add your server to claude_desktop_config.json:

{
  "mcpServers": {
    "task-manager": {
      "command": "uv",
      "args": ["--directory", "/path/to/your/server", "run", "server.py"]
    }
  }
}

Test with the MCP Inspector

mcp dev server.py

This opens a web UI where you can test tools, resources, and prompts interactively.

Production Considerations

Error Handling

Wrap tool implementations with proper error handling:

@mcp.tool()
async def safe_create_task(title: str) -> str:
    """Create a task with error handling."""
    try:
        # Validation
        if not title.strip():
            return json.dumps({"error": "Title cannot be empty"})
        
        if len(title) > 200:
            return json.dumps({"error": "Title too long (max 200 chars)"})
        
        # Create task
        task_id = f"task-{len(tasks) + 1}"
        tasks[task_id] = {"id": task_id, "title": title, "status": "pending"}
        
        return json.dumps({"success": True, "task_id": task_id})
    
    except Exception as e:
        return json.dumps({"error": f"Failed to create task: {str(e)}"})

Logging

Use stderr for logging (stdout is reserved for MCP protocol messages):

import logging
import sys

logging.basicConfig(
    stream=sys.stderr,
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

@mcp.tool()
async def logged_tool(query: str) -> str:
    """A tool with proper logging."""
    logger.info(f"Received query: {query}")
    # ... tool implementation
    logger.info("Tool execution completed")
    return "result"

Transport Selection

Transport	Use Case	Pros	Cons
stdio	Desktop apps (Claude)	Simple, secure	Single client
HTTP	Web servers, APIs	Multi-client, scalable	Requires auth
SSE	Real-time updates	Streaming support	Complex setup

Authentication for HTTP Transport

from mcp.server.auth import BearerTokenAuth

mcp = MCPServer(
    "secure-server",
    auth=BearerTokenAuth(token="your-secret-token")
)

Deploying to Production

Docker

FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install "mcp[cli]"
EXPOSE 8080
CMD ["mcp", "run", "server.py", "--transport", "http", "--port", "8080"]

Environment Variables

import os

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///tasks.db")
API_KEY = os.getenv("API_KEY")
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")

Connecting Multiple MCP Servers

MCP clients can connect to multiple servers simultaneously:

{
  "mcpServers": {
    "task-manager": {
      "command": "uv",
      "args": ["--directory", "/path/to/tasks", "run", "server.py"]
    },
    "calendar": {
      "command": "uv",
      "args": ["--directory", "/path/to/calendar", "run", "server.py"]
    },
    "email": {
      "command": "uv",
      "args": ["--directory", "/path/to/email", "run", "server.py"]
    }
  }
}

The AI agent discovers all tools from all servers and can use them together in complex workflows.

Conclusion

Building an MCP server with the Python SDK is straightforward — define tools with decorators, add resources and prompts, and run the server. The protocol handles discovery, serialization, and transport. Start with stdio for development, add HTTP for production, and layer in authentication and error handling as you scale.