Build an MCP Client (Core)
Build an MCP Client (Core)
This page extracts the core content from the official tutorial, focusing on the minimal viable implementation (Plan A). All examples and code are sourced from the official docs for accuracy.
Recommended prerequisite: /docs/quickstart/server/ (understand the client-server communication model)
System Requirements (Python)
- macOS or Windows
- Latest Python version
uv
installed
Initialize the Project (Python)
# Create project directory
uv init mcp-client
cd mcp-client
# Create virtual environment
uv venv
# Activate virtual environment (macOS/Linux)
source .venv/bin/activate
# Install dependencies
uv add mcp anthropic python-dotenv
# Cleanup boilerplate and create main file
rm main.py
touch client.py
Configure API Key (Anthropic)
echo "ANTHROPIC_API_KEY=<your key here>" > .env
echo ".env" >> .gitignore
Client Skeleton (from official tutorial)
import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from anthropic import Anthropic
from dotenv import load_dotenv
load_dotenv()
class MCPClient:
def __init__(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.anthropic = Anthropic()
# methods will go here
Connect to an MCP server (official example)
async def connect_to_server(self, server_script_path: str):
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
if not (is_python or is_js):
raise ValueError("Server script must be a .py or .js file")
command = "python" if is_python else "node"
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
response = await self.session.list_tools()
tools = response.tools
print("\nConnected to server with tools:", [tool.name for tool in tools])
Query processing (official snippet)
async def process_query(self, query: str) -> str:
messages = [{"role": "user", "content": query}]
response = await self.session.list_tools()
available_tools = [{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
} for tool in response.tools]
response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=messages,
tools=available_tools
)
# Follow the official tutorial for handling tool calls and assembling final output
Next Steps
- Add tool execution and result aggregation
- Add a TypeScript version mirroring the official tutorial
- Add error handling and retry logic
- Integrate OAuth tokens aligned with /specification/draft/basic/authorization/