6.59 · MCP clients — Communication bridge

Mở đầu: Client không phải end-user UI

Your app architecture:

User UI
   │
   ▼
Your server / backend ← app code
   │
   ├─ Anthropic client (call Claude)
   └─ MCP client (call MCP servers)
      │
      ▼
MCP servers (GitHub, Slack, custom, ...)

MCP client = 1 component backend của bạn.

Transport agnostic

MCP không care communication method. Options:

stdio (common)

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

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

Use cases:

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

HTTP

Server runs on remote machine, client connects via HTTP.

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

Use cases:

• Cloud-hosted servers
• Multi-tenant
• Web apps

WebSocket

Persistent bi-directional connection.

Use cases:

• Real-time updates from server to client
• Long-running servers

Message types

MCP spec defines specific request/response pairs:

ListToolsRequest / ListToolsResult

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

CallToolRequest / CallToolResult

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

ListPromptsRequest / ListPromptsResult

Similar cho prompts.

GetPromptRequest / GetPromptResult

Fetch specific prompt template.

ListResourcesRequest / ListResourcesResult

ReadResourceRequest / ReadResourceResult

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

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

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

Python MCP client SDK

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

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)


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"})

Clean API. Auto cleanup.

Multiple MCP servers

App thường connect nhiều servers:

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)

Nested async with OK. Or manage via collection pattern.

Convert MCP tools → Claude tools

MCP tools → Anthropic API format:

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
    }


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
)

Claude sees MCP tools same as your custom tools.

Handling tool results

tool_call = next(b for b in response.content if b.type == "tool_use")

result = await client.call_tool(tool_call.name, tool_call.input)

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

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

Pattern same as Module 5 tool use.

Error handling

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

Graceful degradation important — MCP server can be unavailable.

Connection lifecycle

clients = await start_mcp_clients()

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

Long-running connections work. Clients reuse session.

Anti-patterns

New client per request

async def handle_request(user_msg):
    async with MCPClient(...) as client:  # new session every 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.

Á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 bài học