Construire un client MCP (Core)
Construire un client MCP (Core)
Cette page extrait le contenu essentiel du tutoriel officiel et se concentre sur l’implémentation minimale viable (Plan A). Tous les exemples et codes proviennent de la documentation officielle.
Pré-requis recommandé : /docs/quickstart/server/ (comprendre le modèle de communication client-serveur)
Prérequis système (Python)
- macOS ou Windows
- Dernière version de Python
uv
installé
Initialiser le projet (Python)
uv init mcp-client
cd mcp-client
uv venv
source .venv/bin/activate
uv add mcp anthropic python-dotenv
rm main.py
touch client.py
Configurer la clé API (Anthropic)
echo "ANTHROPIC_API_KEY=<your key here>" > .env
echo ".env" >> .gitignore
Squelette du client (exemple officiel)
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
Se connecter à un serveur MCP (exemple officiel)
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])
Traitement des requêtes (extrait officiel)
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
)
Prochaines étapes
- Ajouter l’exécution d’outils et l’agrégation des résultats
- Ajouter une version TypeScript alignée sur le tutoriel officiel
- Ajouter la gestion d’erreurs et la logique de retry
- Intégrer les jetons OAuth alignés avec /specification/draft/basic/authorization/