7.6 · Triển khai MCP Client đầu tay

Mở đầu: Cái thiếu giữa bạn và Claude

   USER ──► YOUR APP ──► MCP CLIENT ──► MCP SERVER ──► EXTERNAL API
                         ↑
                         │
                   [CHƯA có code]

Đến giờ, chúng ta đã:

• Bài 7.4: build MCP Server với tools
• Bài 7.5: test server bằng Inspector (một MCP client đặc biệt UI-based)

Nhưng để app thật của bạn nói chuyện với server, bạn cần một MCP Client "programmatic" — một class Python có thể gọi trong code, quản lý connection, và expose method đơn giản như list_tools() / call_tool().

Trong bài này, bạn sẽ viết class đó.

Đây cũng là điểm quan trọng: trong dự án thực tế, bạn thường xuyên chỉ xây một phía — hoặc là client, hoặc là server, không phải cả hai. Ví dụ: team ở Anthropic xây Claude Desktop (client) nhưng không tự viết server, họ connect tới server do vendor khác phát hành. Ngược lại, team GitHub xây server mcp.github.com nhưng không viết client, họ chỉ expose protocol.

Khóa học này build cả hai để bạn thấy bức tranh đầy đủ. Nhưng trong sự nghiệp, bạn sẽ chọn một phía để chuyên sâu.

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

┌──────────────────────────────────────────────────────┐
│                                                      │
│   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

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:

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

1. API verbose — session.list_tools() return ListToolsResult với nhiều wrapping. Bạn sẽ gõ .tools nhiều lần khắp codebase.

1. Khó mock — Unit test bạn muốn mock list_tools() — wrap vào class của mình dễ mock hơn.

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.

Code: MCPClient class

Tạo file mcp_client.py:

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

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

Constructor __init__

• 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

connect() — heart of the setup

• 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

session() — safe accessor

• Guard chống truy cập session chưa initialize
• Raise error rõ ràng thay vì để NoneType error ở nơi khác

Tool methods

• list_tools() — simple wrap, đi sâu vào result.tools
• call_tool() — pass-through với type hint

cleanup() + context manager

• AsyncExitStack.aclose() tự cleanup mọi context manager reversed order
• __aenter__ / __aexit__ cho async with syntax

Sử dụng MCPClient — Test riêng

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

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

Chạy:

uv run mcp_client.py

Output:

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

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

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

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

uv run main.py

Thử hỏi:

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.

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

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ẽ:

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

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.

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

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.

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:

ps aux | grep mcp_server

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

Bước 2: Log tool result

Thêm print:

result = await mcp.call_tool(block.name, block.input)
print(f"[DEBUG] Tool result: {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:

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

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

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

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

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:

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

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:

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

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

Data — Analyst notebook

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

Pattern:

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

result = await client.call_tool("run_sql", {"query": "SELECT ..."})

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:

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

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

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

Không dùng async with

Sai lầm:

client = MCPClient(...)
await client.connect()

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:

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

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 chat_loop(mcp):
    user_input = input(">")  # sync, block
    response = requests.post(...)  # sync

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:

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

Tại sao là sai

Anthropic API expect string, result.content là list objects.

Cách đúng

Extract text:

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

Skip error handling

Sai lầm:

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

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:

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

Hard-code transport

Sai lầm:

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

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:

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

Mẹo nâng cao

Mẹo 1: Retry với exponential backoff

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

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

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

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)

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

response = claude.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "..."}],
    mcp_servers=[{
        "type": "url",
        "url": "https://mcp.github.com",
        "name": "github"
    }]
)

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ì:

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

Á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:

uv run mcp_client.py

Bước 4

Verify output list tool match với Inspector.

Bước 5

Ghi lại:

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

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:

uv run main.py

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

• "What's in deposition.md?"
• "Replace 'testimony' with 'statement' in deposition.md."
• "Now show me the updated deposition.md."

Bước 5

Verify Claude chain tool đúng: read → edit → read.

Bước 6

Ghi lại:

• Claude gọi đúng tool đầu tiên (read_doc_contents) không? ___________
• Sequence đúng cho 3 câu? ___________
• Tool call nào unexpected? ___________

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:

@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()
    ]

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

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

Tài liệu tham khảo

• MCP Python SDK — Client usage
• asyncio AsyncExitStack
• Anthropic tool use docs
• MCP connector in Claude API