Triển khai MCP client

Full MCP Client implementation

Clean API cover tools, prompts, resources.

# mcp_client.py
from contextlib import AsyncExitStack
from typing import Any
import mcp.types as types
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


class MCPClient:
    """Wrap MCP ClientSession với auto cleanup."""
    
    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 connect(self):
        """Manual connect (nếu không dùng async with)."""
        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()
    
    async def close(self):
        """Manual close."""
        await self._exit_stack.aclose()
    
    async def __aenter__(self):
        await self.connect()
        return self
    
    async def __aexit__(self, *args):
        await self.close()
    
    def session(self) -> ClientSession:
        if self._session is None:
            raise RuntimeError("Client not connected. Call connect() first.")
        return self._session
    
    async def list_tools(self) -> list[types.Tool]:
        """Get available tools from MCP server."""
        result = await self.session().list_tools()
        return result.tools
    
    async def call_tool(
        self,
        tool_name: str,
        tool_input: dict
    ) -> types.CallToolResult:
        """Execute tool with given arguments."""
        return await self.session().call_tool(tool_name, tool_input)
    
    async def list_prompts(self) -> list[types.Prompt]:
        return (await self.session().list_prompts()).prompts
    
    async def get_prompt(self, name: str, args: dict = None) -> types.GetPromptResult:
        return await self.session().get_prompt(name, args or {})
    
    async def list_resources(self) -> list[types.Resource]:
        return (await self.session().list_resources()).resources
    
    async def read_resource(self, uri: str) -> types.ReadResourceResult:
        return await self.session().read_resource(uri)

Test client standalone

Before wire up to chatbot, test client alone:

Run: uv run test_client.py. Expected:

# test_client.py
import asyncio
from mcp_client import MCPClient


async def main():
    async with MCPClient(
        command="uv",
        args=["run", "mcp_server.py"]
    ) as client:
        # Test list_tools
        tools = await client.list_tools()
        print(f"Found {len(tools)} tools:")
        for tool in tools:
            print(f"  - {tool.name}: {tool.description}")
        
        # Test call_tool (assume hello tool exists)
        result = await client.call_tool("hello", {"name": "Jimmy"})
        print(f"\nTool result: {result.content}")


if __name__ == "__main__":
    asyncio.run(main())

Test client standalone (tiếp)

Found 1 tools:
  - hello: Simple hello tool for testing

Tool result: [TextContent(type='text', text='Hello, Jimmy!')]

Integrate client vào conversation loop

Full main.py:

import asyncio
import json
from dotenv import load_dotenv
from anthropic import Anthropic
from mcp_client import MCPClient

load_dotenv()

anthropic = Anthropic()
MODEL = "claude-sonnet-5-20260205"


def tool_to_anthropic_schema(tool):
    return {
        "name": tool.name,
        "description": tool.description,
        "input_schema": tool.inputSchema
    }


def format_mcp_result(result) -> str:
    """Extract text from MCP CallToolResult."""
    parts = []
    for content in result.content:
        if content.type == "text":
            parts.append(content.text)
        elif content.type == "image":
            parts.append("[image content]")  # handle if needed
    return "\n".join(parts)


async def run_conversation(client: MCPClient):
    tools = await client.list_tools()
    anthropic_tools = [tool_to_anthropic_schema(t) for t in tools]
    
    messages = []
    
    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break
        if not user_input:
            continue
        
        messages.append({"role": "user", "content": user_input})
        
        # Tool use loop
        for turn in range(10):  # max 10 tool turns
            response = anthropic.messages.create(
                model=MODEL,
                max_tokens=2000,
                messages=messages,
                tools=anthropic_tools
            )
            
            messages.append({"role": "assistant", "content": response.content})
            
            # Text blocks (display)
            for block in response.content:
                if block.type == "text" and block.text:
                    print(f"Claude: {block.text}")
            
            if response.stop_reason == "end_turn":
                break
            
            if response.stop_reason == "tool_use":
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        print(f"  → MCP call: {block.name}({json.dumps(block.input)})")
                        try:
                            result = await client.call_tool(block.name, block.input)
                            result_text = format_mcp_result(result)
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": result_text,
                                "is_error": result.isError
                            })
                            print(f"  ✓ Result: {result_text[:100]}...")
                        except Exception as e:
                            tool_results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": str(e),
                                "is_error": True
                            })
                            print(f"  ✗ Error: {e}")
                
                messages.append({"role": "user", "content": tool_results})
                continue
            
            break  # other stop reasons
        
        print()


async def main():
    async with MCPClient(
        command="uv",
        args=["run", "mcp_server.py"]
    ) as client:
        print("Chatbot with MCP ready. Type 'quit' to exit.\n")
        await run_conversation(client)


if __name__ == "__main__":
    asyncio.run(main())

Test end-to-end

uv run main.py

Test end-to-end (tiếp)

Flow working. Now expand with real tools.

Chatbot with MCP ready. Type 'quit' to exit.

You: Use the hello tool to greet Anna
  → MCP call: hello({"name": "Anna"})
  ✓ Result: Hello, Anna!
Claude: Greeted Anna using the hello tool.

You: quit

Error patterns

Connection lost

Tool raises error

try:
    result = await client.call_tool(name, args)
except Exception as e:
    if "connection" in str(e).lower():
        # Reconnect
        await client.connect()
        result = await client.call_tool(name, args)
    else:
        raise

Tool raises error

Timeout

result = await client.call_tool(name, args)
if result.isError:
    # Report to Claude so it can retry
    error_text = format_mcp_result(result)
    logger.warning(f"Tool {name} errored: {error_text}")

Timeout

import asyncio

try:
    result = await asyncio.wait_for(
        client.call_tool(name, args),
        timeout=30.0
    )
except asyncio.TimeoutError:
    logger.error(f"Tool {name} timed out")

Session reuse

Cho app long-running, khởi tạo 1 lần:

class App:
    def __init__(self):
        self._clients: dict[str, MCPClient] = {}
    
    async def start(self):
        self._clients["github"] = MCPClient("github-mcp-server", [])
        await self._clients["github"].connect()
        
        self._clients["slack"] = MCPClient("slack-mcp-server", [])
        await self._clients["slack"].connect()
    
    async def handle_request(self, user_msg):
        # ... use self._clients
        pass
    
    async def shutdown(self):
        for client in self._clients.values():
            await client.close()

Anti-patterns

❌ Connect mỗi request

High overhead, slow first response.

Fix: Connect at app start, reuse.

❌ Không handle isError

Treat error results as valid → confuse downstream.

Fix: Check result.isError, propagate.

❌ Sync chờ trong async context

Block event loop.

Fix: Ensure all ops async.

Áp dụng ngay

Bài tập 1: Test client (15 phút)

Create test_client.py. Run. Verify hello tool works.

Bài tập 2: Add quick math tool (20 phút)

Server: add add_numbers(a, b) tool. Test end-to-end:

• User: "What's 123 + 456?"
• Claude calls tool
• Response: 579

Tóm tắt

🎯 MCP client wrapper với AsyncExitStack cleanup.

🎯 list_tools + call_tool = minimum needed. Add prompts/resources as needed.

🎯 Test client standalone trước khi wire up Claude.

🎯 Session reuse cho long-running apps.

🎯 Handle errors — isError, timeouts, connection loss.