Triển khai MCP Client đầu tay

Kiến trúc: Custom Client vs SDK Session

Tại sao wrap ClientSession?

SDK cung cấp ClientSession — một class low-level giao tiếp với server. Nó hoạt động nhưng có 3 vấn đề nếu dùng trực tiếp trong app logic:

Giải pháp: viết MCPClient class wrap session + expose async context manager + clean methods. Pattern này giống repository pattern trong backend dev.

• Resource management phức tạp — Phải manually setup stdio client, gọi initialize(), đảm bảo aclose() đúng lúc. Quên một bước → zombie process.
• API verbose — session.list_tools() return ListToolsResult với nhiều wrapping. Bạn sẽ gõ .tools nhiều lần khắp codebase.
• Khó mock — Unit test bạn muốn mock list_tools() — wrap vào class của mình dễ mock hơn.

┌──────────────────────────────────────────────────────┐
│                                                      │
│   YOUR APPLICATION (main.py)                         │
│                                                      │
│   ┌──────────────────────────────────────────────┐   │
│   │                                              │   │
│   │       MCPClient (mcp_client.py)              │   │
│   │       ├── __aenter__ / __aexit__             │   │
│   │       ├── list_tools()                       │   │
│   │       ├── call_tool()                        │   │
│   │       ├── read_resource()       (Bài 7.8)    │   │
│   │       ├── list_prompts()        (Bài 7.10)   │   │
│   │       └── get_prompt()          (Bài 7.10)   │   │
│   │                                              │   │
│   │   ┌──────────────────────────────────┐       │   │
│   │   │                                  │       │   │
│   │   │  ClientSession (from mcp SDK)    │       │   │
│   │   │  — protocol primitive            │       │   │
│   │   │                                  │       │   │
│   │   └──────────────────────────────────┘       │   │
│   │                                              │   │
│   └──────────────────────────────────────────────┘   │
│                            ▲                         │
└────────────────────────────┼─────────────────────────┘
                             │ stdio
                             ▼
                      MCP Server process

Code: MCPClient class

Tạo file mcp_client.py:

Bóc tách từng phần

Constructor __init__

connect() — heart of the setup

session() — safe accessor

Tool methods

cleanup() + context manager

• Nhận command + args + env để biết chạy MCP server như thế nào (ví dụ "python", ["mcp_server.py"])
• Chưa start server ở đây — lazy initialization
• StdioServerParameters(...) — config cho stdio transport
• stdio_client(...) — context manager từ SDK mở subprocess + pipe
• AsyncExitStack — track nhiều async context manager để cleanup đúng thứ tự
• ClientSession(_stdio, _write) — wrap 2 file descriptor thành session
• initialize() — gửi InitializeRequest tới server, nhận capabilities
• Guard chống truy cập session chưa initialize
• Raise error rõ ràng thay vì để NoneType error ở nơi khác
• list_tools() — simple wrap, đi sâu vào result.tools
• call_tool() — pass-through với type hint
• AsyncExitStack.aclose() tự cleanup mọi context manager reversed order
• __aenter__ / __aexit__ cho async with syntax

import sys
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client


class MCPClient:
    def __init__(self, command: str, args: list[str], env: Optional[dict] = None):
        self._command = command
        self._args = args
        self._env = env
        self._session: Optional[ClientSession] = None
        self._exit_stack: AsyncExitStack = AsyncExitStack()

    async def connect(self):
        """Start the MCP server subprocess and initialize the session."""
        server_params = StdioServerParameters(
            command=self._command,
            args=self._args,
            env=self._env,
        )
        stdio_transport = await self._exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        _stdio, _write = stdio_transport
        self._session = await self._exit_stack.enter_async_context(
            ClientSession(_stdio, _write)
        )
        await self._session.initialize()

    def session(self) -> ClientSession:
        if self._session is None:
            raise ConnectionError(
                "Client session not initialized. Call connect() first."
            )
        return self._session

    # -------- Tool methods --------

    async def list_tools(self) -> list[types.Tool]:
        result = await self.session().list_tools()
        return result.tools

    async def call_tool(
        self, tool_name: str, tool_input: dict
    ) -> types.CallToolResult | None:
        return await self.session().call_tool(tool_name, tool_input)

    # -------- Lifecycle --------

    async def cleanup(self):
        await self._exit_stack.aclose()
        self._session = None

    async def __aenter__(self):
        await self.connect()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.cleanup()

Sử dụng MCPClient — Test riêng

Thêm harness dưới cùng mcp_client.py:

Chạy:

import asyncio

async def main():
    async with MCPClient(
        command="uv",
        args=["run", "mcp_server.py"],
    ) as client:
        tools = await client.list_tools()
        for t in tools:
            print(f"- {t.name}: {t.description[:50]}...")


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

Sử dụng MCPClient — Test riêng (tiếp)

Output:

uv run mcp_client.py

Sử dụng MCPClient — Test riêng (tiếp)

Thấy list tool → client đã connect và query server thành công.

- read_doc_contents: Read the full text contents of a document by...
- edit_document: Edit a document by replacing exact text with...

Ghép vào CLI chat app

Bây giờ ghép MCPClient + Anthropic SDK thành CLI chat hoàn chỉnh.

File: main.py

Chạy

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

load_dotenv()


def tools_for_claude(mcp_tools):
    """Convert MCP Tool objects to Anthropic API tool format."""
    return [
        {
            "name": tool.name,
            "description": tool.description,
            "input_schema": tool.inputSchema,
        }
        for tool in mcp_tools
    ]


async def chat_loop(claude: Anthropic, mcp: MCPClient):
    model = os.environ.get("CLAUDE_MODEL", "claude-sonnet-5")
    mcp_tools = await mcp.list_tools()
    tools = tools_for_claude(mcp_tools)

    messages = []

    while True:
        user_input = input("\nBạn > ").strip()
        if user_input.lower() in {"quit", "exit"}:
            break

        messages.append({"role": "user", "content": user_input})

        # Agentic loop: giữ gọi Claude cho đến khi không còn tool_use
        while True:
            response = claude.messages.create(
                model=model,
                max_tokens=4096,
                tools=tools,
                messages=messages,
            )

            # Thêm response của Claude vào messages
            messages.append({
                "role": "assistant",
                "content": response.content,
            })

            if response.stop_reason != "tool_use":
                # Xong, in câu trả lời cuối
                for block in response.content:
                    if block.type == "text":
                        print(f"\nClaude > {block.text}")
                break

            # Có tool_use: chạy tất cả và thêm kết quả
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"\n[Tool call: {block.name}({block.input})]")
                    result = await mcp.call_tool(block.name, block.input)
                    content_text = ""
                    if result and result.content:
                        for c in result.content:
                            if hasattr(c, "text"):
                                content_text += c.text
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": content_text,
                        "is_error": bool(result and result.isError),
                    })

            messages.append({
                "role": "user",
                "content": tool_results,
            })


async def main():
    claude = Anthropic()

    async with MCPClient(
        command="uv",
        args=["run", "mcp_server.py"],
    ) as mcp:
        print("MCP connected. Gõ 'quit' để thoát.")
        await chat_loop(claude, mcp)


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

Chạy

Thử hỏi:

uv run main.py

Ghép vào CLI chat app (tiếp)

Bạn vừa có một mini-Claude Desktop — tự viết, chạy local, tool thật của bạn.

Bạn > What's in the plan.md document?

[Tool call: read_doc_contents({'doc_id': 'plan.md'})]

Claude > The plan.md document outlines the steps for the project's
         implementation.

Flow end-to-end — Xem lại với code thật

Bây giờ đối chiếu 11 bước từ Bài 7.2 với code thực tế:

Flow trong bài 7.2 giờ đã chạy thật. Mọi bước đều có code mapping trực tiếp.

Agentic loop — Tại sao vòng while?

Một câu hỏi của user có thể trigger nhiều tool call liên tiếp.

Ví dụ: "Summarize all documents."

Claude sẽ:

Nếu client code chỉ gọi Claude một lần, sau bước 1 sẽ thấy stop_reason = "tool_use", thực thi tool, rồi kết thúc. Claude không có cơ hội gọi tool tiếp.

Pattern while True chạy cho đến khi Claude không còn muốn gọi tool nữa (stop_reason = "end_turn"). Đây là agentic loop cơ bản.

Advanced production setup bạn còn thêm: max iterations, cost tracking, streaming, error recovery, tool timeout. Course này focus pattern cơ bản — phần nâng cao ở các course khác.

• Gọi list_documents() → biết có 6 docs
• Gọi read_doc_contents("deposition.md")
• Gọi read_doc_contents("report.pdf")
• ... (6 lần)
• Cuối cùng generate text summary

┌───────────────────────────────────────────┐
│                                           │
│   user_msg                                │
│    │                                      │
│    ▼                                      │
│   Claude API                              │
│    │                                      │
│    ▼                                      │
│   stop_reason?                            │
│    │                                      │
│    ├── "end_turn" ───► Print + break      │
│    │                                      │
│    └── "tool_use"                         │
│         │                                 │
│         ▼                                 │
│        Run tools via MCP                  │
│         │                                 │
│         ▼                                 │
│        Append tool results                │
│         │                                 │
│         └──────► (loop back to Claude)    │
│                                           │
└───────────────────────────────────────────┘

Bảng so sánh: 3 cách gọi MCP từ code

Bài này dạy cách 2 vì nó balance giữa học hỏi vs tiện dụng. Cách 3 (mcp_servers param trực tiếp trong claude.messages.create()) là production shortcut khi app bạn SaaS + không cần control chi tiết loop — sẽ nhắc qua ở Bài 7.11.

Ví dụ thực chiến: Debug một tool_use loop treo

Tình huống

Bạn chạy main.py, hỏi một câu. Terminal in [Tool call: ...], rồi treo. Không có error, không có response.

Bước 1: Check server subprocess

Terminal khác:

Server process đang chạy? Còn responding?

Bước 2: Log tool result

Thêm print:

ps aux | grep mcp_server

Bước 2: Log tool result

Thấy gì? None? Empty? Error?

Bước 3: Check timeout

Tool của bạn có block > 60s không? Thêm timeout:

result = await mcp.call_tool(block.name, block.input)
print(f"[DEBUG] Tool result: {result}")

Bước 3: Check timeout

Bước 4: Check messages format

Log messages trước khi gửi Claude:

try:
    result = await asyncio.wait_for(
        mcp.call_tool(block.name, block.input),
        timeout=30.0
    )
except asyncio.TimeoutError:
    print("Tool timed out!")

Bước 4: Check messages format

Format đúng chưa? tool_result block có tool_use_id match với tool_use tương ứng?

Kết quả

90% "tool_use loop treo" đi qua 4 check này. Thường là:

• Tool raise exception chưa catch → result.isError=True nhưng content empty
• tool_use_id không match → Claude confused
• Tool serialize object Python không JSON-able → serialize fail silent

import json
print(f"[DEBUG] {json.dumps(messages, default=str, indent=2)}")

Ví dụ theo ngành — Custom Client patterns

🛠️ Internal tools — VS Code extension

Context: Team build VS Code extension kết nối nhiều MCP server internal (Jira, Git, Internal docs).

Pattern:

Kết quả: 1 client manage 5+ MCP servers cùng lúc. User không thấy complexity.

💼 SaaS — Chat widget trên web

Context: B2B SaaS cho customer support. Customer chat với AI, backend gọi MCP server vendor có OAuth.

Pattern:

Kết quả: Scale horizontally, cost thấp.

📊 Data — Analyst notebook

Context: Data analyst dùng Jupyter + MCP để query database, visualize.

Pattern:

• Remote HTTP transport (không stdio)
• Per-user session (mỗi customer conn riêng)
• Cache tool list (tool không đổi thường xuyên → cache 5 phút)

class MultiMCPClient:
    def __init__(self):
        self.clients = {}

    async def connect_all(self, configs):
        for name, cfg in configs.items():
            client = MCPClient(**cfg)
            await client.connect()
            self.clients[name] = client

    async def list_all_tools(self):
        # Tool từ mọi server, prefix name để tránh conflict
        all_tools = []
        for name, client in self.clients.items():
            tools = await client.list_tools()
            for t in tools:
                t.name = f"{name}_{t.name}"
            all_tools.extend(tools)
        return all_tools

📊 Data — Analyst notebook

Kết quả: No context manager vì notebook cell-based. Analyst quen async trong Jupyter.

🎓 Student projects — Quick prototype

Context: Hackathon project, 24h deadline.

Pattern: Copy MCPClient template, add 1-2 tool cụ thể, skip polish. Focus demo.

Kết quả: Working prototype trong 3-4h. Tech debt accepted cho demo.

🏢 Enterprise — Audited production

Context: Bank tích hợp Claude với internal systems, compliance audit.

Pattern:

Kết quả: Pass audit. Tool fail gracefully không downtime.

• MCPClient mở rộng thêm: retry, circuit breaker, metrics export (Prometheus)
• Every call_tool() log structured JSON vào SIEM
• Timeout aggressive (10s default)
• Tool result truncation (max 10KB mỗi call)

# notebook cell 1
client = MCPClient(command="uv", args=["run", "sql_mcp.py"])
await client.connect()
tools = await client.list_tools()

# notebook cell 2+: dùng client trực tiếp
result = await client.call_tool("run_sql", {"query": "SELECT ..."})

Anti-patterns — Những sai lầm cần tránh

❌ Không dùng async with

Sai lầm:

Tại sao là sai: Subprocess orphan. Memory leak. Zombie process tích lũy theo thời gian.

Cách đúng: Luôn dùng async with MCPClient(...) as client:.

❌ Mỗi request mở client mới

Sai lầm:

client = MCPClient(...)
await client.connect()
# ... code ...
# quên await client.cleanup()

❌ Mỗi request mở client mới

Tại sao là sai: Mỗi request spawn subprocess mới (~100-500ms). User feels lag.

Cách đúng: Mở 1 client ở app startup, reuse cho mọi request.

❌ Block event loop với sync call

Sai lầm:

async def handle_request(user_msg):
    async with MCPClient(...) as client:
        ...

❌ Block event loop với sync call

Tại sao là sai: input() và requests block event loop. Không async workload khác chạy được.

Cách đúng: Dùng asyncio.to_thread(input, ">") và httpx.AsyncClient cho HTTP.

❌ Nhét tool result dạng object raw

Sai lầm:

async def chat_loop(mcp):
    user_input = input(">")  # sync, block
    response = requests.post(...)  # sync

❌ Nhét tool result dạng object raw

Tại sao là sai: Anthropic API expect string, result.content là list objects.

Cách đúng: Extract text:

tool_results.append({
    "type": "tool_result",
    "tool_use_id": block.id,
    "content": result.content,  # list of TextContent objects!
})

Anti-patterns — Những sai lầm cần tránh (tiếp)

❌ Skip error handling

Sai lầm:

content_text = ""
for c in result.content:
    if hasattr(c, "text"):
        content_text += c.text

❌ Skip error handling

Tại sao là sai: Tool error → result.isError=True → content chứa error string. Pass vào Claude làm Claude confused.

Cách đúng: Mark tool_result với is_error: True khi isError:

result = await mcp.call_tool(...)
return result.content[0].text  # crash nếu isError=true

Anti-patterns — Những sai lầm cần tránh (tiếp)

❌ Hard-code transport

Sai lầm:

tool_results.append({
    ...,
    "is_error": bool(result.isError),
})

❌ Hard-code transport

Tại sao là sai: Không portable. Khi deploy cần đổi sang uv run hoặc remote HTTP.

Cách đúng: Load config từ file:

client = MCPClient(command="python", args=["server.py"])

Anti-patterns — Những sai lầm cần tránh (tiếp)

config = load_config("mcp.json")
client = MCPClient(**config)

Mẹo nâng cao

Mẹo 1: Retry với exponential backoff

Mẹo 2: Metrics decorator

async def call_tool_retry(self, name, input, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await self.session().call_tool(name, input)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)

Mẹo 2: Metrics decorator

Mẹo 3: Auto-discovery tool từ nhiều server

Khi bạn có 3+ MCP server, mỗi cái ở file config riêng:

import time

def timed(method):
    async def wrapper(self, *args, **kwargs):
        start = time.perf_counter()
        try:
            return await method(self, *args, **kwargs)
        finally:
            elapsed = time.perf_counter() - start
            print(f"[METRIC] {method.__name__}: {elapsed:.2f}s")
    return wrapper

class MCPClient:
    @timed
    async def call_tool(self, ...): ...

Mẹo 3: Auto-discovery tool từ nhiều server

Mẹo 4: MCP connector trong API (alternative)

Production shortcut không cần tự viết agentic loop:

import json
from pathlib import Path

def load_mcp_configs():
    config_dir = Path.home() / ".mcp" / "servers"
    return [
        json.loads(f.read_text())
        for f in config_dir.glob("*.json")
    ]

Mẹo 4: MCP connector trong API (alternative)

Anthropic API tự handle loop. Code bạn giảm ~60 dòng. Dùng khi app SaaS + không cần custom loop.

Nhưng bài này dạy manual loop vì:

• Học hiểu flow sâu hơn
• Con đường tới custom logic (guard, monitoring, custom error handling)
• Chạy local stdio MCP server (API connector chỉ remote)

# Không cần MCPClient!
response = claude.messages.create(
    model="claude-sonnet-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "..."}],
    mcp_servers=[{
        "type": "url",
        "url": "https://mcp.github.com",
        "name": "github"
    }]
)

Áp dụng ngay

Bài tập 1: Build MCPClient (~20 phút)

Bước 1: Tạo mcp_client.py trong project. Copy class MCPClient từ bài.

Bước 2: Thêm test harness main() cuối file.

Bước 3: Chạy:

Bước 4: Verify output list tool match với Inspector.

Bước 5: Ghi lại:

Bài tập 2: CLI chat đầu tay (~20 phút)

Bước 1: Tạo main.py copy từ bài.

Bước 2: Verify .env có ANTHROPIC_API_KEY (từ Bài 7.3).

Bước 3: Chạy:

• Số tool list ra: ___________
• Thời gian từ start tới in tool: ___________
• Có lỗi nào không? ___________

uv run mcp_client.py

Bài tập 2: CLI chat đầu tay (~20 phút)

Bước 4: Thử 3 câu hỏi:

Bước 5: Verify Claude chain tool đúng: read → edit → read.

Bước 6: Ghi lại:

Bài tập 3 (thử thách): Thêm tool thứ 3 (~15 phút)

Quay lại mcp_server.py, thêm tool list_documents:

• "What's in deposition.md?"
• "Replace 'testimony' with 'statement' in deposition.md."
• "Now show me the updated deposition.md."
• Claude gọi đúng tool đầu tiên (read_doc_contents) không? ___________
• Sequence đúng cho 3 câu? ___________
• Tool call nào unexpected? ___________

uv run main.py

Bài tập 3 (thử thách): Thêm tool thứ 3 (~15 phút)

Restart main.py, thử:

"Show me all documents I have."

Verify:

• Claude gọi list_documents không? ___________
• Output format readable? ___________
• Bạn có cần sửa code main.py không? Tại sao? (Hint: không — tool mới auto available vì list_tools() call dynamic) ___________

@mcp.tool(
    name="list_documents",
    description="List all available document IDs with a short preview..."
)
def list_documents():
    return [
        {"id": k, "preview": v[:50] + "..."}
        for k, v in docs.items()
    ]

Tóm tắt bài học

🎯 MCPClient là wrapper custom quanh SDK ClientSession — Lifecycle management, clean API, easy to mock. Pattern ai cũng nên biết.

🎯 async with là mandatory — AsyncExitStack + context manager đảm bảo cleanup. Không dùng = zombie subprocess.

🎯 2 method cốt lõi — list_tools() trả list tool, call_tool(name, input) execute. Thêm method cho resources/prompts ở bài sau.

🎯 Agentic loop là while stop_reason == "tool_use" — Claude có thể gọi tool nhiều lần. Code phải loop cho đến khi end_turn.

🎯 Tool_result format precise — tool_use_id match, content là string, is_error boolean. Một field sai → Claude confused.