Build an MCP client in FastAPI

Introduction

When you wire an MCP server into a FastAPI app and let every incoming request spawn a fresh subprocess to reach it, you pay 200–500ms of startup latency per call and risk leaking orphaned server processes when handlers crash mid-tool-call — silently turning a working prototype into a production outage. By the end of this lesson, you will be able to build an MCPToolClient that initializes a Model Context Protocol session in the app's lifespan, exposes list_tools() and call_tool() to request handlers via dependency injection, and handles the concurrency, timeout, and health-check concerns that keep the integration safe under load.

Key Terminology

• MCP client: The application-side component that opens a session to an MCP server, performs tools/list and tools/call JSON-RPC exchanges, and surfaces results to the LLM or request handler.
• ClientSession: The mcp SDK class that owns the JSON-RPC protocol state — request IDs, pending futures, initialize handshake — over a pair of read/write streams supplied by a transport.
• stdio transport: A transport that spawns the MCP server as a subprocess and exchanges newline-delimited JSON-RPC messages over its stdin/stdout pipes; configured via StdioServerParameters and entered with stdio_client().

Concepts

Why the client matters more than the server

Most tutorials spend 90% of their time on the server side — decorating functions with @mcp.tool() and writing JSON schema descriptions. In production, the server is the easy part. The client is where complexity lives because it must handle transport negotiation, session lifecycle management, concurrent tool calls from multiple request handlers, and graceful degradation when a server becomes unreachable. A poorly designed client becomes a bottleneck: if every FastAPI request spawns a new subprocess to connect to an MCP server, you pay 200–500ms of process startup latency per call. If you share a single session across requests without synchronization, you corrupt the JSON-RPC message stream. The MCPToolClient pattern we build here solves both problems by maintaining a persistent session pool with proper async locking.

Production considerations

Connection pooling across multiple servers. Real applications connect to multiple MCP servers — one for document search, one for database queries, one for external APIs. Extend MCPToolClient into an MCPToolRouter that maintains a dictionary of {server_name: MCPToolClient} instances, each initialized in the lifespan. The list_tools() method aggregates tools across all servers, prefixing names with the server identifier to avoid collisions (e.g., docs__search_docs vs db__run_query).

Concurrency safety. The ClientSession uses sequential JSON-RPC message IDs over a single stream. If two FastAPI request handlers call call_tool() concurrently on the same session, their responses can get interleaved. Protect the session with an asyncio.Lock or — better — maintain a pool of sessions and check them out per-request.

Timeout and retry. MCP tool calls can hang if the server process deadlocks or the underlying service is slow. Wrap every call_tool() invocation in asyncio.wait_for(client.call_tool(...), timeout=30.0) and catch asyncio.TimeoutError to return a graceful error message to the LLM rather than blocking the entire request.

Health checks. Add a /health/mcp endpoint that calls list_tools() and returns the count. If the session has died (server process crashed), catch the exception and return a 503 status, allowing your load balancer to route traffic away from unhealthy instances while the lifespan restarts the connection.

Code Walkthrough

MCP client-server interaction model

Before writing code, you need a precise mental model of how the client communicates with MCP servers during a single LLM generation request. The client does not maintain a permanent open connection by default — it initializes a session, performs operations, and the session persists for the duration of the logical interaction.

Loading diagram...

This diagram captures two critical phases. First, tool discovery happens once per session — the client sends a tools/list JSON-RPC request and receives back an array of tool definitions, each containing a name, description, and inputSchema (a JSON Schema object describing required and optional parameters). Second, tool invocation happens zero or more times — the client sends a tools/call request with the tool name and a dictionary of arguments, and receives back a CallToolResult containing one or more content blocks (text, images, or embedded resources). The transport layer — either stdio pipes to a subprocess or an SSE HTTP connection — handles serialization transparently.

Building the MCPToolClient class

The following implementation creates an MCPToolClient class in services/mcp_client.py that wraps the mcp SDK's ClientSession with lifecycle management suitable for FastAPI. The class uses mcp.client.stdio.stdio_client to spawn a server subprocess and establish a session through mcp.client.session.ClientSession. It exposes two primary methods: list_tools() for discovery and call_tool() for invocation. The aenter and aexit dunder methods enable use as an async context manager, ensuring the transport and session are properly initialized and torn down. Pay particular attention to how the StdioServerParameters dataclass configures the server command — in production, this would come from a configuration file or environment variable rather than being hardcoded.

1import asyncio
2from dataclasses import dataclass
3from typing import Any
4from mcp.client.stdio import stdio_client, StdioServerParameters
5from mcp.client.session import ClientSession
6from mcp.types import Tool, CallToolResult, TextContent
7
8@dataclass
9class ToolDefinition:
10    """LLM-friendly tool definition extracted from MCP Tool objects."""
11    name: str
12    description: str
13    parameters: dict[str, Any]  # JSON Schema from inputSchema
14
15class MCPToolClient:
16    """Async MCP client that manages session lifecycle for FastAPI."""
17
18    def __init__(self, server_command: str, server_args: list[str] | None = None):
19        self._params = StdioServerParameters(
20            command=server_command,
21            args=server_args or [],
22            env=None,
23        )
24        self._session: ClientSession | None = None
25        self._transport_ctx = None
26        self._session_ctx = None
27        self._read_stream = None
28        self._write_stream = None
29
30    async def __aenter__(self) -> "MCPToolClient":
31        self._transport_ctx = stdio_client(self._params)
32        streams = await self._transport_ctx.__aenter__()
33        self._read_stream, self._write_stream = streams
34        self._session_ctx = ClientSession(
35            self._read_stream, self._write_stream
36        )
37        self._session = await self._session_ctx.__aenter__()
38        await self._session.initialize()
39        return self
40
41    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
42        if self._session_ctx:
43            await self._session_ctx.__aexit__(exc_type, exc_val, exc_tb)
44        if self._transport_ctx:
45            await self._transport_ctx.__aexit__(exc_type, exc_val, exc_tb)
46
47    async def list_tools(self) -> list[ToolDefinition]:
48        if self._session is None:
49            raise RuntimeError("Client not initialized — use async with")
50        result = await self._session.list_tools()
51        return [
52            ToolDefinition(
53                name=tool.name,
54                description=tool.description or "",
55                parameters=tool.inputSchema,
56            )
57            for tool in result.tools
58        ]
59
60    async def call_tool(
61        self, name: str, arguments: dict[str, Any]
62    ) -> str:
63        if self._session is None:
64            raise RuntimeError("Client not initialized — use async with")
65        result: CallToolResult = await self._session.call_tool(
66            name, arguments
67        )
68        if result.isError:
69            texts = [c.text for c in result.content if isinstance(c, TextContent)]
70            raise RuntimeError(f"Tool {name} failed: {'; '.join(texts)}")
71        return "\n".join(
72            c.text for c in result.content if isinstance(c, TextContent)
73        )

Integrating the client into FastAPI with dependency injection

The MCPToolClient must be available to your route handlers without creating a new subprocess connection per request. The following code shows how to register the client as a FastAPI lifespan dependency, creating the session once at application startup and sharing it across all requests. The get_mcp_client dependency function uses FastAPI's Depends mechanism to inject the initialized client into any endpoint that needs tool access. The lifespan async context manager — defined using the @asynccontextmanager decorator — ensures the client session is properly closed when the application shuts down, preventing orphaned MCP server subprocesses.

1from contextlib import asynccontextmanager
2from fastapi import FastAPI, Depends, Request
3from services.mcp_client import MCPToolClient, ToolDefinition
4
5_mcp_client: MCPToolClient | None = None
6
7@asynccontextmanager
8async def lifespan(app: FastAPI):
9    global _mcp_client
10    client = MCPToolClient(
11        server_command="python",
12        server_args=["-m", "tools.mcp_server"],
13    )
14    _mcp_client = await client.__aenter__()
15    yield
16    await client.__aexit__(None, None, None)
17
18app = FastAPI(lifespan=lifespan)
19
20async def get_mcp_client() -> MCPToolClient:
21    if _mcp_client is None:
22        raise RuntimeError("MCP client not initialized")
23    return _mcp_client
24
25@app.get("/tools")
26async def list_available_tools(
27    client: MCPToolClient = Depends(get_mcp_client),
28) -> list[ToolDefinition]:
29    return await client.list_tools()
30
31@app.post("/tools/{tool_name}/invoke")
32async def invoke_tool(
33    tool_name: str,
34    request: Request,
35    client: MCPToolClient = Depends(get_mcp_client),
36) -> dict:
37    body = await request.json()
38    arguments = body.get("arguments", {})
39    result_text = await client.call_tool(tool_name, arguments)
40    return {"tool": tool_name, "result": result_text}

Connecting discovery to LLM function calling

Tool discovery is only half the story. The ToolDefinition objects returned by list_tools() must be translated into the format your LLM provider expects — but because the MCP inputSchema is already a JSON Schema dictionary, the mapping is mechanical: wrap each definition as {"type": "function", "function": {"name": tool.name, "description": tool.description, "parameters": tool.parameters}} and feed the resulting list to your chat-completion call's tools argument. The same shape works for OpenAI, and for Anthropic and Gemini via LiteLLM — no manual schema rewriting.

When the LLM responds with a tool_calls array, you extract each call's function.name and function.arguments (a JSON string that you parse), then dispatch them through client.call_tool(name, arguments). The string result goes back into the conversation as a message with role: "tool". This dispatch loop — which a full agentic executor expands with iteration limits and error recovery — is the mechanism that closes the feedback loop between the LLM's reasoning and your MCP server's business logic. You'll know the integration works end-to-end when GET /tools returns the server's tool catalog and POST /tools/{name}/invoke returns a non-empty result string for a known-valid argument set.

Do's and Don'ts

Do's

1. ✓Do initialize MCPToolClient once in the FastAPI app lifespan, not per request — each call to stdio_client(StdioServerParameters(...)) spawns a new server subprocess, adding 200–500ms of startup latency per handler invocation and leaving orphaned processes whenever a handler crashes before __aexit__ runs.
2. ✓Do tear down _session_ctx before _transport_ctx in __aexit__ — the ClientSession must send its JSON-RPC shutdown messages before the underlying stdio streams are closed; reversing the order corrupts the close handshake and leaves the MCP server subprocess hanging without a clean exit.
3. ✓Do raise RuntimeError in list_tools() and call_tool() when self._session is None — this surfaces a misconfigured dependency injection path immediately at the call site rather than letting an AttributeError on None obscure that MCPToolClient was never entered as an async context manager.

Don'ts

1. ✗Don't skip checking result.isError on CallToolResult before reading result.content — MCP tool failures arrive as structurally valid JSON-RPC responses with isError=True, not as Python exceptions; bypassing the check silently forwards error text to the LLM as though it were legitimate tool output, producing hallucinated downstream reasoning.
2. ✗Don't hardcode the server command in StdioServerParameters — the lesson flags this as an explicit production anti-pattern; the command and args fields must come from a configuration file or environment variable so the same MCPToolClient class works across dev, staging, and production without source changes.
3. ✗Don't expose raw mcp.types.Tool objects directly to FastAPI request handlers — wrap them in the ToolDefinition dataclass (extracting only name, description, and parameters from inputSchema) so handlers are decoupled from the MCP SDK's internal type hierarchy and survive SDK version upgrades without cascading changes.