MCP clients — Communication bridge

Transport agnostic

MCP không care communication method. Options:

stdio (common)

Client và server run same machine, communicate via stdin/stdout.

Use cases:

HTTP

Server runs on remote machine, client connects via HTTP.

• Local dev
• Server is local executable
• Desktop apps (Claude Desktop)

[Client process] ──stdin──▶ [Server process]
                 ◀─stdout──

HTTP

Use cases:

WebSocket

Persistent bi-directional connection.

Use cases:

• Cloud-hosted servers
• Multi-tenant
• Web apps
• Real-time updates from server to client
• Long-running servers

[Client] ──HTTP POST──▶ [Server]
         ◀─HTTP Response──

Message types

MCP spec defines specific request/response pairs:

ListToolsRequest / ListToolsResult

CallToolRequest / CallToolResult

Client: "What tools do you provide?"
Server: [{name: "get_repos", ...}, {name: "create_issue", ...}]

CallToolRequest / CallToolResult

ListPromptsRequest / ListPromptsResult

Similar cho prompts.

GetPromptRequest / GetPromptResult

Fetch specific prompt template.

ListResourcesRequest / ListResourcesResult

ReadResourceRequest / ReadResourceResult

Client: "Call get_repos with {owner: 'anthropics'}"
Server: "Tool result: [repo1, repo2, ...]"

Full flow example: "What repos do I have?"

Many steps, each with clear responsibility. Scale qua các tools dễ.

1. User: "What repos do I have?" → Your app
    
2. Your app: needs to tell Claude about GitHub tools
    → MCP client: ListToolsRequest
    → GitHub MCP server: replies [get_repos, list_prs, ...]
    ← MCP client: list of tool schemas
    
3. Your app: call Claude with user msg + tools
    → Anthropic API
    ← Claude response: ToolUseBlock "call get_repos()"
    
4. Your app: execute tool via MCP client
    → MCP client: CallToolRequest("get_repos", {})
    → GitHub MCP server: calls GitHub REST API
    ← GitHub API: [repo1, repo2, ...]
    ← MCP server: CallToolResult
    ← MCP client: tool result
    
5. Your app: send result back to Claude
    → Anthropic API with tool_result
    ← Claude: "You have 12 repos: ..."
    
6. Your app: forward to User UI

Python MCP client SDK

Minimal example (stdio)

pip install mcp

Minimal example (stdio)

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main():
    # Connect to MCP server (executable)
    server_params = StdioServerParameters(
        command="python",
        args=["github_mcp_server.py"],
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize
            await session.initialize()
            
            # List tools
            tools_result = await session.list_tools()
            print("Available tools:", [t.name for t in tools_result.tools])
            
            # Call tool
            result = await session.call_tool("get_repos", {"owner": "anthropics"})
            print("Result:", result.content)


asyncio.run(main())

Wrapper class for convenience

Clean API. Auto cleanup.

class MCPClient:
    def __init__(self, command: str, args: list[str]):
        self.command = command
        self.args = args
        self._session: ClientSession | None = None
        self._exit_stack = AsyncExitStack()
    
    async def __aenter__(self):
        params = StdioServerParameters(command=self.command, args=self.args)
        read, write = await self._exit_stack.enter_async_context(stdio_client(params))
        self._session = await self._exit_stack.enter_async_context(
            ClientSession(read, write)
        )
        await self._session.initialize()
        return self
    
    async def __aexit__(self, *args):
        await self._exit_stack.aclose()
    
    async def list_tools(self):
        return (await self._session.list_tools()).tools
    
    async def call_tool(self, name: str, args: dict):
        return await self._session.call_tool(name, args)


# Usage
async def main():
    async with MCPClient("python", ["server.py"]) as client:
        tools = await client.list_tools()
        result = await client.call_tool("my_tool", {"arg": "value"})

Multiple MCP servers

App thường connect nhiều servers:

Nested async with OK. Or manage via collection pattern.

async def main():
    async with MCPClient("python", ["github_server.py"]) as github:
        async with MCPClient("python", ["slack_server.py"]) as slack:
            # Get tools from both
            github_tools = await github.list_tools()
            slack_tools = await slack.list_tools()
            
            all_tools = github_tools + slack_tools
            # ... pass to Claude
            
            # Dispatch: depending tool name, route to correct MCP client
            tool_call = ... # từ Claude response
            if tool_call.name in [t.name for t in github_tools]:
                result = await github.call_tool(tool_call.name, tool_call.input)
            elif tool_call.name in [t.name for t in slack_tools]:
                result = await slack.call_tool(tool_call.name, tool_call.input)

Convert MCP tools → Claude tools

MCP tools → Anthropic API format:

Claude sees MCP tools same as your custom tools.

def mcp_tool_to_anthropic_schema(mcp_tool):
    """Convert MCP tool definition to Anthropic tool schema."""
    return {
        "name": mcp_tool.name,
        "description": mcp_tool.description,
        "input_schema": mcp_tool.inputSchema
    }


# Usage
mcp_tools = await client.list_tools()
anthropic_tools = [mcp_tool_to_anthropic_schema(t) for t in mcp_tools]

response = anthropic.messages.create(
    model=MODEL,
    max_tokens=1000,
    messages=messages,
    tools=anthropic_tools
)

Handling tool results

Pattern same as Module 5 tool use.

# Claude responds with ToolUseBlock
tool_call = next(b for b in response.content if b.type == "tool_use")

# Execute via MCP
result = await client.call_tool(tool_call.name, tool_call.input)

# Extract text content
result_text = ""
for content in result.content:
    if content.type == "text":
        result_text += content.text

# Send back to Claude
messages.append({
    "role": "user",
    "content": [{
        "type": "tool_result",
        "tool_use_id": tool_call.id,
        "content": result_text,
        "is_error": result.isError  # MCP spec
    }]
})

Error handling

Graceful degradation important — MCP server can be unavailable.

try:
    result = await client.call_tool(name, args)
    if result.isError:
        error_text = result.content[0].text if result.content else "Unknown error"
        raise MCPToolError(error_text)
    return result
except Exception as e:
    logger.error(f"MCP tool call failed: {e}")
    # Send error to Claude so it can handle
    ...

Connection lifecycle

Long-running connections work. Clients reuse session.

# App startup
clients = await start_mcp_clients()

# App runtime
async def handle_request(user_msg):
    all_tools = gather_tools_from_all_clients(clients)
    # ... serve request
    
# App shutdown
for client in clients:
    await client.close()

Anti-patterns

❌ New client per request

Overhead high (connect + initialize).

Fix: Session pool, reuse across requests.

❌ Ignore tool name collision

GitHub MCP get_user and Slack MCP get_user — same name.

Fix: Namespace github:get_user, slack:get_user.

❌ No timeout

MCP server hangs → app stuck.

Fix: Timeout wrapping call_tool.

❌ Trust MCP server output blindly

Security risk if server untrusted.

Fix: Validate output types, whitelist allowed servers.

async def handle_request(user_msg):
    async with MCPClient(...) as client:  # new session every request
        ...

Áp dụng ngay

Bài tập 1: Connect to public MCP server (30 phút)

Install Filesystem MCP server. Connect via your client.

List tools, call read_file tool. Verify works.

Bài tập 2: Multi-server setup (30 phút)

Connect 2 MCP servers simultaneously. List all tools combined. Dispatch tool calls correctly.

Tóm tắt

🎯 MCP client = bridge giữa app và MCP servers.

🎯 Transport agnostic: stdio, HTTP, WebSocket.

🎯 Message types: ListTools, CallTool, ListPrompts, ListResources.

🎯 Connect multiple servers, merge tools, dispatch by name.

🎯 Session reuse qua requests cho performance.