---

name: mcp-client-integration description: Integrates MCP clients using Python SDK v2 and TypeScript SDK v2 to connect to MCP servers, manage tools/resources/prompts, handle transport (stdio/SSE), error recovery, and implement structured calling conventions in AI agent applications. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding role: implementation scope: implementation output-format: code triggers: mcp client, mcp integration, how do i connect to mcp, consuming mcp servers, claude mcp client, typescript mcp, tool invocation related-skills: mcp-server-fastmcp-python, mcp-tool-design-patterns archetypes: tactical anti_triggers: brainstorming, vague ideation response_profile: verbosity: low directive_strength: high abstraction_level: operational

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).

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.

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

async def invoke_tool_bad(tool_name: str, args: dict) -> dict:
    """Dangerous: Silently fails, no clear error context."""
    try:
        result = await session.call_tool(tool_name, args)
        return result.content
    except:
        # What error? What tool failed? No context.
        return {}  # Silent failure — caller has no idea what went wrong
    
    # No timeout protection → can hang indefinitely
    # No retry logic → transient failures break the agent
    # No validation → malformed responses corrupt downstream logic

Problems:

• Empty dict is indistinguishable from successful "no result" case
• Caller has no information about what failed or why
• No timeout means a slow/hanging server blocks the entire agent
• No retry for transient failures (network hiccups, server restart)

✅ GOOD: Explicit Error Types and Context

class ToolInvocationError(Exception):
    """Base exception for tool invocation failures."""
    def __init__(self, tool_name: str, reason: str, is_retryable: bool = False):
        self.tool_name = tool_name
        self.reason = reason
        self.is_retryable = is_retryable
        super().__init__(f"Tool '{tool_name}' failed: {reason}")


async def invoke_tool_good(
    tool_name: str,
    arguments: dict,
    max_retries: int = 3,
    timeout_seconds: float = 30.0
) -> dict:
    """Correct: Clear errors, retry logic, timeout protection.
    
    Args:
        tool_name: Name of tool to invoke
        arguments: Input arguments for tool
        max_retries: Number of retries for transient errors
        timeout_seconds: Maximum time for single invocation
    
    Returns:
        Tool result dictionary
    
    Raises:
        ToolInvocationError: With clear reason and retryability flag
        ValueError: If arguments don't match tool schema
    """
    if tool_name not in tools_cache:
        raise ValueError(
            f"Tool '{tool_name}' not found in available tools: {list(tools_cache.keys())}"
        )
    
    last_error = None
    
    for attempt in range(1, max_retries + 1):
        try:
            # Timeout protection: prevents hanging on slow servers
            result = await asyncio.wait_for(
                session.call_tool(tool_name, arguments),
                timeout=timeout_seconds
            )
            
            # Validate response structure
            if not hasattr(result, "content"):
                raise ToolInvocationError(
                    tool_name,
                    "Malformed response: missing 'content' field",
                    is_retryable=False
                )
            
            return result.content
        
        except asyncio.TimeoutError:
            last_error = ToolInvocationError(
                tool_name,
                f"Timeout after {timeout_seconds}s",
                is_retryable=True
            )
        except ConnectionError as e:
            last_error = ToolInvocationError(
                tool_name,
                f"Connection error: {e}",
                is_retryable=True
            )
        except ValueError as e:
            # Schema validation error — not retryable
            raise ToolInvocationError(
                tool_name,
                f"Invalid arguments: {e}",
                is_retryable=False
            )
        except Exception as e:
            last_error = ToolInvocationError(
                tool_name,
                f"Unexpected error: {type(e).__name__}: {e}",
                is_retryable=False
            )
        
        # If not retryable, fail immediately
        if last_error and not last_error.is_retryable:
            raise last_error
        
        # Exponential backoff before retry
        if attempt < max_retries:
            backoff_seconds = 2 ** (attempt - 1)  # 1s, 2s, 4s, ...
            print(f"Attempt {attempt}/{max_retries} failed, retrying in {backoff_seconds}s: {last_error.reason}")
            await asyncio.sleep(backoff_seconds)
    
    # All retries exhausted
    raise last_error or ToolInvocationError(
        tool_name,
        f"Failed after {max_retries} retries",
        is_retryable=False
    )


# Usage with proper error handling
try:
    result = await invoke_tool_good("list_files", {"directory": "/tmp"})
    print(f"Success: {result}")
except ToolInvocationError as e:
    if e.is_retryable:
        print(f"Transient error (safe to retry): {e.reason}")
        # Agent can decide to retry or backoff
    else:
        print(f"Permanent error (don't retry): {e.reason}")
        # Agent should fail immediately

Improvements:

• Explicit error types with is_retryable flag for agent decision-making
• Timeout protection prevents hanging
• Exponential backoff reduces server load on transient failures
• Clear context in error messages (tool name, reason, suggestion)
• Validates response structure before returning
• Distinguishes transient (timeout, connection) from permanent (validation, not found) errors

---

Error Recovery Patterns

Handling Timeouts

import asyncio

async def invoke_with_timeout(tool_name: str, args: dict, timeout_sec: float = 30.0) -> dict:
    """Invoke tool with timeout and graceful timeout handling."""
    try:
        result = await asyncio.wait_for(
            session.call_tool(tool_name, args),
            timeout=timeout_sec
        )
        return result.content
    except asyncio.TimeoutError:
        raise TimeoutError(
            f"Tool '{tool_name}' did not respond within {timeout_sec}s. "
            f"Server may be overloaded or hung. Consider increasing timeout or retrying."
        )

Handling Tool Not Found

async def safe_invoke_tool(tool_name: str, args: dict) -> dict:
    """Invoke tool with validation that it exists first."""
    available_tools = {t.name for t in (await session.list_tools()).tools}
    
    if tool_name not in available_tools:
        raise ValueError(
            f"Tool '{tool_name}' not found. Available tools: {', '.join(sorted(available_tools))}"
        )
    
    return (await session.call_tool(tool_name, args)).content

Handling Malformed Responses

import json

async def invoke_with_validation(tool_name: str, args: dict) -> dict:
    """Invoke tool and validate response structure."""
    result = await session.call_tool(tool_name, args)
    
    # Validate response has required fields
    if not isinstance(result.content, (list, dict, str)):
        raise ValueError(
            f"Malformed response from '{tool_name}': "
            f"expected dict/list/str, got {type(result.content).__name__}"
        )
    
    # Try to parse if JSON string
    if isinstance(result.content, str):
        try:
            return json.loads(result.content)
        except json.JSONDecodeError as e:
            raise ValueError(
                f"Tool '{tool_name}' returned invalid JSON: {e}"
            )
    
    return result.content

---

Transport Layer Details

Stdio Transport (Local Servers)

Best for:

• Local development
• Private MCP servers (filesystem, database)
• Low-latency requirements
• Subprocess-managed servers

from mcp import StdioServerParameters, ClientSession

# Launch server as subprocess
params = StdioServerParameters(
    command="/usr/local/bin/mcp-server-postgres",
    args=["--database", "production"]
)

session = await ClientSession.create(params)

Advantages: Direct process control, low latency, secure (no network)
Disadvantages: Server must run locally, subprocess management overhead

SSE Transport (HTTP Streaming)

Best for:

• Remote servers (cloud deployments)
• Firewall-friendly (HTTP only)
• Multi-tenant architectures
• Load-balanced server instances

from mcp import ServerParameters
import aiohttp

# Connect to HTTP SSE server
async with aiohttp.ClientSession() as http_session:
    params = ServerParameters(
        url="https://mcp-server.example.com/sse",
        headers={"Authorization": f"Bearer {api_key}"}
    )
    session = await ClientSession.create(params)

Advantages: Remote access, scalable, firewall-friendly
Disadvantages: Network latency, requires server deployment

---

Connection Lifecycle Management

Proper Initialization (Handshake)

async def connect_and_verify(server_path: str) -> ClientSession:
    """Connect with verification that server is responsive."""
    session = await ClientSession.create(StdioServerParameters(command=server_path))
    
    try:
        # Verify server is responsive by listing tools
        tools_response = await asyncio.wait_for(
            session.list_tools(),
            timeout=5.0
        )
        print(f"Connected. Available tools: {[t.name for t in tools_response.tools]}")
        return session
    except Exception as e:
        await session.close()
        raise ConnectionError(f"Server verification failed: {e}")

Graceful Shutdown

async def shutdown_client(session: ClientSession) -> None:
    """Close connection and cleanup resources."""
    try:
        await session.close()
        print("Client closed gracefully")
    except Exception as e:
        print(f"Warning: Error closing client: {e}")
    finally:
        # Ensure cleanup even if close() fails
        session = None

Context Manager Pattern (Recommended)

from contextlib import asynccontextmanager

@asynccontextmanager
async def mcp_client(server_path: str):
    """Context manager ensures proper cleanup."""
    session = None
    try:
        session = await ClientSession.create(StdioServerParameters(command=server_path))
        yield session
    finally:
        if session:
            await session.close()

# Usage
async with mcp_client("/usr/local/bin/mcp-server-filesystem") as session:
    tools = await session.list_tools()
    # Do work...
# Session automatically closed when exiting block

---

Best Practices

1. Cache Tool Metadata

class MCPClient:
    def __init__(self):
        self.tools_cache: dict = {}
        self.cache_timestamp = None
        self.cache_ttl_seconds = 3600  # Refresh every hour
    
    async def get_tools(self, force_refresh: bool = False) -> list:
        """Get tools with caching to reduce server load."""
        import time
        
        now = time.time()
        should_refresh = (
            force_refresh or
            not self.tools_cache or
            (self.cache_timestamp and now - self.cache_timestamp > self.cache_ttl_seconds)
        )
        
        if should_refresh:
            tools = await self.session.list_tools()
            self.tools_cache = {t.name: t for t in tools.tools}
            self.cache_timestamp = now
        
        return list(self.tools_cache.values())

2. Implement Health Checks

async def health_check(session: ClientSession, timeout_sec: float = 5.0) -> bool:
    """Quick check that server is responsive."""
    try:
        await asyncio.wait_for(
            session.list_tools(),
            timeout=timeout_sec
        )
        return True
    except Exception:
        return False

3. Type-Safe Tool Invocation

from typing import TypedDict

class WeatherInput(TypedDict):
    location: str
    unit: str  # "celsius" or "fahrenheit"

async def get_weather(args: WeatherInput) -> dict:
    """Type-safe wrapper for weather tool."""
    return await invoke_tool("get_weather", args)

---

Constraints

MUST DO

• Always use context managers or async cleanup for connection lifecycle management
• Implement timeout protection for all tool invocations (prevent hanging)
• Classify errors as retryable vs permanent to guide agent behavior
• Cache tool/resource metadata to reduce server load
• Validate tool response structure before using results
• Log clear error messages with tool name, reason, and recovery suggestion
• Test with realistic error scenarios (timeouts, missing tools, malformed responses) before production
• Use exponential backoff for retrying transient failures (1s, 2s, 4s, ...)
• Document required environment variables and server endpoint configuration
• Handle connection loss gracefully with automatic reconnection for long-lived clients

---

MUST NOT DO

• Do NOT silently fail with empty results or default values — throw descriptive errors
• Do NOT use generic exception catching without classifying error type
• Do NOT retry permanent errors (invalid arguments, tool not found) — waste time and resources
• Do NOT invoke unbounded retries — cap at 3-5 retries with backoff
• Do NOT block on tool invocation without timeout — can hang the entire agent
• Do NOT assume tools are cached — call list_tools() if unsure
• Do NOT parse malformed tool responses without validation — corrupts downstream logic
• Do NOT share client sessions across async tasks without synchronization
• Do NOT hardcode server paths/URLs — use environment variables and config files
• Do NOT leave connections open on shutdown — implement proper cleanup

---

Resource Discovery and Management

Listing Resources

async def discover_resources(session: ClientSession) -> dict[str, list]:
    """Discover and categorize available resources."""
    response = await session.list_resources()
    
    resources_by_type = {}
    for resource in response.resources:
        resource_type = resource.uri.split("://")[0]  # e.g., "file", "database"
        if resource_type not in resources_by_type:
            resources_by_type[resource_type] = []
        resources_by_type[resource_type].append({
            "uri": resource.uri,
            "name": resource.name,
            "description": resource.description,
            "mime_type": resource.mimeType
        })
    
    return resources_by_type

Reading Resource Content

async def read_resource(session: ClientSession, resource_uri: str) -> str:
    """Read content from a resource (file, database query, API response)."""
    try:
        response = await asyncio.wait_for(
            session.read_resource(resource_uri),
            timeout=10.0
        )
        
        # Handle different response types
        if isinstance(response.contents, str):
            return response.contents
        elif isinstance(response.contents, bytes):
            return response.contents.decode("utf-8")
        else:
            return str(response.contents)
    
    except asyncio.TimeoutError:
        raise TimeoutError(f"Reading resource '{resource_uri}' exceeded timeout")
    except Exception as e:
        raise RuntimeError(f"Failed to read resource '{resource_uri}': {e}")

---

Testing Integration

Unit Test for Tool Invocation

import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.mark.asyncio
async def test_invoke_tool_success():
    """Test successful tool invocation."""
    # Mock session
    mock_session = AsyncMock()
    mock_response = MagicMock()
    mock_response.content = {"result": "success"}
    mock_session.call_tool.return_value = mock_response
    
    # Test invocation
    result = await invoke_tool_good("test_tool", {"arg": "value"})
    assert result == {"result": "success"}
    mock_session.call_tool.assert_called_once_with("test_tool", {"arg": "value"})


@pytest.mark.asyncio
async def test_invoke_tool_not_found():
    """Test error when tool doesn't exist."""
    with pytest.raises(ValueError, match="Tool 'missing_tool' not found"):
        await invoke_tool_good("missing_tool", {})


@pytest.mark.asyncio
async def test_invoke_tool_timeout():
    """Test timeout handling and retry logic."""
    mock_session = AsyncMock()
    mock_session.call_tool.side_effect = asyncio.TimeoutError()
    
    with pytest.raises(ToolInvocationError) as exc_info:
        await invoke_tool_good("slow_tool", {}, max_retries=1, timeout_seconds=0.1)
    
    assert "Timeout" in str(exc_info.value)

---

Common Integration Patterns

Pattern: Agent with Tool Use Loop

async def agent_loop_with_tools(user_message: str, max_iterations: int = 10) -> str:
    """Run agent loop that invokes tools until completion."""
    messages = [{"role": "user", "content": user_message}]
    
    for iteration in range(max_iterations):
        # Get Claude response
        response = await client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=[tool.to_claude_format() for tool in available_tools],
            messages=messages
        )
        
        if response.stop_reason == "end_turn":
            # Claude finished
            return extract_text(response)
        
        if response.stop_reason == "tool_use":
            # Process tool calls
            for block in response.content:
                if block.type == "tool_use":
                    try:
                        result = await invoke_tool_good(block.name, block.input)
                        messages.append({"role": "assistant", "content": response.content})
                        messages.append({
                            "role": "user",
                            "content": [{
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": json.dumps(result)
                            }]
                        })
                    except ToolInvocationError as e:
                        # Return error to Claude
                        messages.append({
                            "role": "user",
                            "content": [{
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "is_error": True,
                                "content": e.reason
                            }]
                        })
    
    return "Max iterations reached"

---

Related Documentation

• MCP Protocol Specification: https://modelcontextprotocol.io
• Claude SDK for Python: https://github.com/anthropics/anthropic-sdk-python
• Claude SDK for TypeScript: https://github.com/anthropics/anthropic-sdk-typescript
• MCP Server Pattern: See mcp-server-fastmcp-python skill
• Tool Design: See mcp-tool-design-patterns skill