Building an MCP Server in Python
Source: mortalapps.com- An MCP server is a standardized interface that exposes tools, resources, and prompts to LLM clients via the Model Context Protocol.
- It solves the problem of fragmented tool integration by providing a universal, transport-agnostic schema for agentic capabilities.
- In production, MCP servers enable a 'write once, run anywhere' architecture for enterprise data connectors and internal APIs.
- This tutorial results in a production-ready Python server supporting both stdio and Streamable HTTP (current recommended transport; SSE is a separate legacy/deprecated transport) transports.
Why This Matters
To build mcp server python components is to adopt a paradigm shift in how AI agents interact with external systems. Historically, developers were forced to write bespoke integration layers for every LLM framework - one for LangChain, another for Semantic Kernel, and a third for custom internal bots. This fragmentation created significant technical debt and maintenance overhead. The Model Context Protocol (MCP) was invented to decouple the 'intelligence' of the model from the 'capability' of the tools. By standardizing the communication layer, a single Python-based MCP server can simultaneously serve a local Claude Desktop instance, a cloud-hosted agentic workflow, and a fleet of autonomous IDE agents.
Ignoring this standard in production leads to 'integration silos' where valuable enterprise data is trapped behind proprietary wrappers that cannot be easily audited, secured, or reused across different model providers. From a production engineering perspective, MCP provides a structured way to enforce schema validation, rate limiting, and security boundaries at the edge of the tool execution environment. It allows teams to move away from granting LLMs broad API access and instead provide granular, well-defined functions. This approach is essential when building systems that must comply with strict governance frameworks, as it centralizes the logic for tool execution, logging, and access control into a single, testable service. Use this approach when you need to expose internal databases, legacy APIs, or specialized compute functions to any MCP-compliant agent without rewriting the interface for each new model or framework version.
Core Concepts
MCP Server Components
An MCP server is composed of several logical layers that handle the lifecycle of a request from an LLM client:
- Transport Layer: The physical communication medium. Common types include
stdio(standard input/output) for local processes andSSE(Server-Sent Events) for networked HTTP services. - Protocol Handler: The logic that parses JSON-RPC 2.0 messages, manages session state, and handles capability negotiation during the
initializehandshake. - Tools: Executable functions that the LLM can invoke. Each tool must have a name, a description (used by the LLM for discovery), and a strict input schema.
- Resources: Passive data sources (like files or database rows) that the LLM can read but not execute.
- Prompts: Pre-defined templates that help the client structure interactions with the server's specific capabilities.
Transport Variants
| Transport | Use Case | Latency | Security Model |
|---|---|---|---|
| stdio | Local IDE extensions, CLI tools | Ultra-low | Process-level isolation |
| SSE (HTTP) | Remote agents, multi-tenant apps | Moderate | TLS, OAuth 2.1, API Keys |
How It Works
The MCP Lifecycle
- Initialization: The client (e.g., Claude Desktop) starts the server process or connects via HTTP. They exchange
initializemessages to negotiate capabilities. The server declares it supports tools and resources; the client declares its protocol version. - Discovery: The client sends a
tools/listrequest. The server responds with a list of available tools, including their Pydantic-generated JSON schemas and descriptions. This is how the LLM 'knows' what it can do. - The Call-Response Loop: When the LLM decides to use a tool, the client sends a
tools/callrequest containing the tool name and arguments. The server validates these arguments against the schema before execution. - Execution and Error Handling: The server executes the Python function. If the function fails, the server must catch the exception and return a structured JSON-RPC error object rather than crashing. This allows the LLM to 'see' the error and potentially self-correct.
- Shutdown: The client sends a
notifications/cancelledor simply closes the transport. The server must perform a clean exit, closing any open database connections or file handles.
Architecture
The architecture follows a classic Hub-and-Spoke model where the MCP Server acts as the spoke providing specialized capabilities to a central LLM Hub. Execution starts at the Client (Hub), which initiates a JSON-RPC 2.0 connection over a Transport (stdio or SSE). The message flows into the MCP Python SDK, which acts as a dispatcher. The dispatcher routes the request to the specific Tool Handler based on the method name. Data flows from the Tool Handler to external systems (Databases, APIs) and back. The result is wrapped in a Content object (Text, Image, or Resource) and sent back through the Transport to the Client. The system is designed to be stateless at the protocol level, though the underlying Python logic may maintain its own stateful connections to external resources.
Environment Setup and SDK Selection
To build mcp server python applications, the primary dependency is the mcp Python SDK. This library provides high-level abstractions for both the low-level JSON-RPC logic and the transport implementations. Start by creating a virtual environment and installing the core package along with uvicorn for HTTP support.
pip install mcp[cli] uvicorn pydanticDefining Tools with Type Safety
The most critical part of an MCP server is the tool definition. The LLM relies entirely on the tool's description and parameter schema to understand when and how to call it. Using the FastMCP class simplifies this by automatically generating schemas from Python type hints and docstrings.
Implementing the Stdio Transport
For local development, stdio is the default. The server listens on stdin and writes to stdout. It is vital to redirect all application logs to stderr to avoid polluting the protocol stream. If your code prints a debug statement to stdout, the MCP client will fail to parse the resulting JSON-RPC message, leading to a connection drop.
Transitioning to Streamable HTTP (current recommended transport; SSE is a separate legacy/deprecated transport)
In production environments, you often need to host the MCP server as a web service. MCP uses Server-Sent Events (SSE) for the server-to-client stream and standard POST requests for client-to-server messages. This 'Streamable HTTP' approach allows for long-running tool executions without the overhead of full WebSockets.
Schema Validation and Constraints
Every tool input should be validated using Pydantic models. This prevents 'prompt injection' style attacks where an LLM might try to pass malformed data to your internal systems. By defining Field(..., ge=0, le=100) or regex patterns in your Pydantic models, you provide the LLM with hard constraints that improve the reliability of the agentic loop.
Handling Large Contexts and Resources
Resources in MCP allow you to expose large datasets without forcing the LLM to 'call' a tool. For example, a resource could be a URI like db://logs/today. When the client requests this resource, the server streams the content. This is more efficient than tool calls for bulk data transfer. Use the list_resources and read_resource handlers to implement this. Ensure you implement pagination for resources that could return megabytes of data, as most LLM context windows are still a bottleneck.
Code Example
import os
import logging
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server
mcp = FastMCP("WeatherService")
# Configure logging to stderr to avoid breaking stdio protocol
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@mcp.tool()
def get_weather(city: str) -> str:
"""Get the current weather for a specific city."""
# In production, use os.getenv for API keys
api_key = os.getenv("WEATHER_API_KEY")
if not api_key:
return "Error: API key not configured."
logger.info(f"Fetching weather for {city}")
# Mocking API call for demonstration
return f"The weather in {city} is sunny, 22°C."
if __name__ == "__main__":
# Run using stdio transport by default
mcp.run()The server starts and waits for JSON-RPC input on stdin. When queried by an MCP client, it returns weather data.
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route
# Create low-level MCP server
server = Server("ProductionServer")
# Define tool handlers manually for granular control
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
if name == "query_db":
return [{"type": "text", "text": "Query results..."}]
raise ValueError(f"Unknown tool: {name}")
# SSE Transport setup
sse = SseServerTransport("/messages")
async def handle_sse(request):
async with sse.connect_scope(request.scope, request.receive, request.send) as scope:
await server.run(
scope.incoming,
scope.outgoing,
server.create_initialization_options()
)
app = Starlette(routes=[
Route("/sse", endpoint=handle_sse),
Route("/messages", endpoint=sse.handle_post_messages, methods=["POST"])
])A web server listening on port 8000, providing an MCP endpoint via SSE for remote agent connections.