mcp-client-integration

# MCP Client Integration Implement MCP clients that connect to MCP servers, discover tools/resources/prompts, handle transport layers (stdio, SSE), and invoke server capabilities with proper error recovery in AI agent applications. ## TL;DR Checklist - [ ] Choose transport type: stdio (local), SSE (HTTP), or custom connection - [ ] Initialize client with proper server configuration and timeout settings - [ ] Implement tool discovery and resource caching mechanisms - [ ] Handle all error cases: timeouts, malformed responses, tool not found - [ ] Close connections gracefully on shutdown (context managers / async cleanup) - [ ] Test tool invocation with realistic error scenarios before production - [ ] Document required environment variables and server endpoints --- ## When to Use Use this skill when: - Building an AI agent that needs to consume tools from external MCP servers - Integrating Claude SDK with MCP clients for tool discovery and invocation - Connecting to local stdio-based servers (e.g., filesystem, database servers) - Consuming HTTP SSE-based MCP servers (streaming transport) - Implementing fallback/retry logic for unreliable MCP server connections - Caching tool/resource metadata to reduce server load --- ## When NOT to Use Avoid this skill for: - Building an MCP server (use `mcp-server-fastmcp-python` instead) - One-off tool calls without agent context (use raw HTTP requests) - Servers that don't implement MCP protocol (use native SDKs) - Synchronous code that can't handle async/await patterns (refactor to async) - Simple shell commands or subprocess calls (use `subprocess` module directly) --- ## Core Workflow ### 1. **Initialize Client with Transport** Choose the appropriate transport layer based on server type: - **Stdio**: Local in-process or subprocess servers - **SSE (HTTP)**: Remote servers, streaming responses - **Custom**: Bidirectional WebSocket or other protocols **Checkpoint:** Server is running, endpoint/executable is accessible, credentials are configured. ### 2. **Discover Available Capabilities** List and cache: - **Tools**: Callable functions with inputs/outputs - **Resources**: Named data sources (files, databases, API endpoints) - **Prompts**: Pre-defined prompt templates **Checkpoint:** Tool catalog is populated, resource URIs are validated. ### 3. **Invoke Tools with Error Handling** Call tools with proper: - Type validation for inputs - Timeout constraints - Error classification (recoverable vs permanent) - Retry logic for transient failures **Checkpoint:** Tool invocation succeeds or raises descriptive error with context. ### 4. **Manage Connection Lifecycle** Maintain connection with: - Graceful initialization (handshake, capability negotiation) - Periodic health checks for long-lived connections - Proper cleanup on shutdown (close, disconnect) - Recovery from connection loss **Checkpoint:** Client can reconnect automatically, resources are freed on exit. --- ## Implementation Patterns ### Pattern 1: Python Client with Stdio Transport Use this for local MCP servers running as subprocesses (e.g., `mcp-server-filesystem`, `mcp-server-postgres`). ```python import asyncio import json from mcp import ClientSession, StdioServerParameters from anthropic import Anthropic class MCPClientManager: """Manage MCP client connection and tool invocation.""" def __init__(self, server_path: str, server_args: list = None): """Initialize stdio-based MCP client. Args: server_path: Path to MCP server executable server_args: Command-line arguments for server Raises: FileNotFoundError: If server executable doesn't exist ValueError: If server_path is empty """ if not server_path: raise ValueError("server_path cannot be empty") self.server_path = server_path self.server_args = server_args or [] self.session: ClientSession = None self.tools_cache: dict = {} async def connect(self) -> None: """Establish connection to MCP server via stdio. Raises: ConnectionError: If server fails to start or handshake fails TimeoutError: If connection takes longer than 10 seconds """ params = StdioServerParameters( command=self.server_path, args=self.server_args ) try: self.session = await asyncio.wait_for( ClientSession.create(params), timeout=10.0 ) except asyncio.TimeoutError: raise TimeoutError(f"MCP server {self.server_path} failed to start within 10s") except Exception as e: raise ConnectionError(f"Failed to connect to MCP server: {e}") async def discover_tools(self) -> list[dict]: """Discover available tools from MCP server. Returns: List of tool definitions with name, description, input schema Raises: RuntimeError: If not connected to server ValueError: If tool discovery fails """ if not self.session: raise RuntimeError("Not connected. Call connect() first.") try: response = await self.session.list_tools() self.tools_cache = {tool.name: tool for tool in response.tools} return [ { "name": tool.name, "description": tool.description, "input_schema": tool.inputSchema } for tool in response.tools ] except Exception as e: raise ValueError(f"Tool discovery failed: {e}") async def invoke_tool(self, tool_name: str, arguments: dict) -> str: """Invoke a tool on the MCP server. Args: tool_name: Name of tool to invoke arguments: Input arguments (must match tool's inputSchema) Returns: Tool result as JSON string Raises: ValueError: If tool not found or arguments invalid TimeoutError: If invocation exceeds 30 seconds RuntimeError: If server returns error response """ if not self.session: raise RuntimeError("Not connected. Call connect() first.") if tool_name not in self.tools_cache: raise ValueError(f"Tool '{tool_name}' not found. Available: {list(self.tools_cache.keys())}") try: result = await asyncio.wait_for( self.session.call_tool(tool_name, arguments), timeout=30.0 ) return json.dumps(result.content) except asyncio.TimeoutError: raise TimeoutError(f"Tool invocation '{tool_name}' exceeded 30s timeout") except Exception as e: raise RuntimeError(f"Tool invocation failed: {tool_name}: {e}") async def close(self) -> None: """Close MCP session gracefully. Raises: RuntimeError: If close fails (logs error, continues shutdown) """ if self.session: try: await self.session.close() except Exception as e: print(f"Warning: Error closing MCP session: {e}") finally: self.session = None async def example_usage(): """Example: Connect to filesystem server and list files.""" manager = MCPClientManager("/usr/local/bin/mcp-server-filesystem") try: await manager.connect() tools = await manager.discover_tools() print(f"Available tools: {[t['name'] for t in tools]}") # Invoke 'list_directory' tool result = await manager.invoke_tool( "list_directory", {"path": "/tmp"} ) print(f"Directory listing: {result}") finally: await manager.close() # Run example if __name__ == "__main__": asyncio.run(example_usage()) ``` --- ### Pattern 2: TypeScript Client with Tool Caching and Retry Use this for integrating MCP clients with Claude SDK in TypeScript agents. ```typescript import Anthropic from "@anthropic-ai/sdk"; interface ToolDefinition { name: string; description: string; input_schema: Record<string, unknown>; } class MCPClientWithRetry { private client: Anthropic; private toolsCache: Map<string, ToolDefinition> = new Map(); private maxRetries: number = 3; private retryDelayMs: number = 1000; constructor(apiKey?: string) { this.client = new Anthropic({ apiKey: apiKey || process.env.ANTHROPIC_API_KEY, }); } /** * Discover and cache tools from MCP server. * Caching reduces server load on repeated agent loops. */ async discoverTools(): Promise<ToolDefinition[]> { // In real implementation, fetch from MCP server // For demo, return mock tools const tools: ToolDefinition[] = [ { name: "get_weather", description: "Get current weather for a location", input_schema: { type: "object", properties: { location: { type: "string", description: "City name", }, }, required: ["location"], }, }, ]; tools.forEach((tool) => this.toolsCache.set(tool.name, tool)); return tools; } /** * Invoke tool with exponential backoff retry logic. * Handles transient failures gracefully. */ async invokeTool( toolName: string, arguments: Record<string, unknown> ): Promise<string> { if (!this.toolsCache.has(toolName)) { throw new Error( `Tool '${toolName}' not found. Available: ${Array.from(this.toolsCache.keys()).join(", ")}` ); } let lastError: Error | null = null; for (let attempt = 1; attempt <= this.maxRetries; attempt++) { try { // In real implementation, call actual MCP server // For demo, simulate tool invocation return await this.simulateToolCall(toolName, arguments); } catch (error) { lastError = error instanceof Error ? error : new Error(String(error)); // Check if error is retryable if (!this.isRetryable(lastError)) { throw lastError; } // Exponential backoff: 1s, 2s, 4s if (attempt < this.maxRetries) { const delayMs = this.retryDelayMs * Math.pow(2, attempt - 1); console.warn( `Tool invocation failed (attempt ${attempt}/${this.maxRetries}), retrying in ${delayMs}ms: ${lastError.message}` ); await new Promise((resolve) => setTimeout(resolve, delayMs)); } } } throw new Error( `Tool invocation '${toolName}' failed after ${this.maxRetries} retries: ${lastError?.message}` ); } /** * Classify errors as retryable (transient) or permanent. */ private isRetryable(error: Error): boolean { const message = error.message.toLowerCase(); return ( message.includes("timeout") || message.includes("econnrefused") || message.includes("econnreset") || message.includes("503") || message.includes("service unavailable") ); } /** * Simulate tool call (replace with actual MCP server invocation). */ private async simulateToolCall( toolName: string, arguments: Record<string, unknown> ): Promise<string> { // Simulate network call await new Promise((resolve) => setTimeout(resolve, 50)); return JSON.stringify({ tool: toolName, input: arguments, result: "Tool executed successfully", }); } /** * Run agent loop with tool use. * Demonstrates integration with Claude SDK. */ async runAgent(userMessage: string): Promise<string> { const tools = await this.discoverTools(); const messages: Anthropic.MessageParam[] = [ { role: "user", content: userMessage, }, ]; // Convert tool definitions to Claude SDK format const claudeTools: Anthropic.Tool[] = tools.map((tool) => ({ name: tool.name, description: tool.description, input_schema: tool.input_schema, })); let response = await this.client.messages.create({ model: "claude-3-5-sonnet-20241022", max_tokens: 1024, tools: claudeTools, messages: messages, }); // Process tool calls in agent loop while (response.stop_reason === "tool_use") { const toolUseBlock = response.content.find( (block) => block.type === "tool_use" ) as Anthropic.ToolUseBlock | undefined; if (!toolUseBlock) break; const toolName = toolUseBlock.name; const toolInput = toolUseBlock.input as Record<string, unknown>; try { const toolResult = await this.invokeTool(toolName, toolInput); // Continue conversation with tool result messages.push({ role: "assistant", content: response.content, }); messages.push({ role: "user", content: [ { type: "tool_result", tool_use_id: toolUseBlock.id, content: toolResult, }, ], }); response = await this.client.messages.create({ model: "claude-3-5-sonnet-20241022", max_tokens: 1024, tools: claudeTools, messages: messages, }); } catch (error) { // Return error to Claude const errorMsg = error instanceof Error ? error.message : String(error); messages.push({ role: "user", content: [ { type: "tool_result", tool_use_id: toolUseBlock.id, is_error: true, content: errorMsg, }, ], }); response = await this.client.messages.create({ model: "claude-3-5-sonnet-20241022", max_tokens: 1024, tools: claudeTools, messages: messages, }); } } // Extract final text response const textBlock = response.content.find((block) => block.type === "text"); return textBlock && "text" in textBlock ? textBlock.text : ""; } } // Usage const client = new MCPClientWithRetry(); client.runAgent("What's the weather in San Francisco?"); ``` --- ### Pattern 3: BAD vs GOOD Error Handling #### ❌ BAD: Silent Failures and Unclear Recovery ```python async def invoke_tool_bad(tool_name: str, args: dict) -> dict: """Dangerous: Silently fails, no clear error context.""" try: