Building MCP Servers

MCP Server Structure

MCP Server components:
1. Server core (handles protocol)
2. Tool definitions (describe capabilities)
3. Tool handlers (implement logic)
4. Transport layer (stdio, HTTP, etc)
5. Authentication (optional)

Creating a Simple MCP Server

from mcp.server import Server
from mcp.types import Tool

server = Server("my-tool-server")

# Define a tool
search_tool = Tool(
    name="web_search",
    description="Search the web for information",
    inputSchema={
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "max_results": {"type": "integer"}
        }
    }
)

# Register tool
@server.tool("web_search")
def search(query: str, max_results: int = 5):
    # Actual implementation here
    return f"Results for {query}"

# Start server
server.run()

Tool Definition Schema

{
  "name": "calculate",
  "description": "Perform mathematical operations",
  "inputSchema": {
    "type": "object",
    "properties": {
      "expression": {
        "type": "string",
        "description": "Math expression to evaluate"
      },
      "precision": {
        "type": "integer",
        "description": "Decimal places (default 2)"
      }
    },
    "required": ["expression"]
  }
}

Tool Validation

Server validates inputs before calling handler:

@server.tool("divide")
def divide(numerator: float, denominator: float) -> float:
    if denominator == 0:
        raise ValueError("Cannot divide by zero")
    return numerator / denominator

# Client calls: divide(10, 0)
# Server validation: Rejects (denominator cannot be 0)
# Client never reaches handler (safe!)

Stateful Servers

Servers can maintain state:

class DatabaseServer(Server):
    def __init__(self):
        super().__init__("db-server")
        self.db_connection = None
    
    def on_startup(self):
        self.db_connection = connect_to_database()
    
    @server.tool("query")
    def query(self, sql: str):
        return self.db_connection.execute(sql)
    
    def on_shutdown(self):
        self.db_connection.close()

Error Handling

@server.tool("fetch_user")
def fetch_user(user_id: int):
    try:
        # Try to fetch user
        user = db.get_user(user_id)
        if not user:
            return {
                "error": "user_not_found",
                "message": f"User {user_id} not found"
            }
        return {"success": True, "user": user}
    
    except DatabaseError as e:
        return {
            "error": "database_error",
            "message": str(e)
        }

Transport Options

1. Stdio (default)
   - Simple, works everywhere
   - Parent process spawns server

2. HTTP
   - Scalable
   - Easy to deploy
   - Add reverse proxy/load balancer

3. WebSocket
   - Bidirectional communication
   - Real-time updates
   - Works in browser