10.7 · Walkthrough — Triển khai notifications end-to-end

Project overview — Cấu trúc thư mục

notifications/
├── .gitignore
├── client.py          # Client CLI nhận notifications
├── pyproject.toml     # Dependency config (uv)
├── README.md          # Hướng dẫn chạy
└── server.py          # Server expose tool với log/progress

Download: link notifications.zip có trong bài 10.6 gốc. Hoặc clone từ repo Anthropic chính thức.

cd ~/mcp-advanced
unzip notifications.zip
cd notifications/
uv sync  # cài dependencies theo pyproject.toml

server.py — Emit notifications

Đây là file quan trọng nhất. Nội dung đầy đủ (với comment để đọc song song):

from mcp.server.fastmcp import FastMCP, Context
from pydantic import Field
import asyncio

mcp = FastMCP(name="notifications-demo")

async def do_research(topic: str) -> list[dict]:
    await asyncio.sleep(2)  # giả lập fetch web
    return [
        {"url": f"https://source1.com/{topic}", "content": "..."},
        {"url": f"https://source2.com/{topic}", "content": "..."},
        {"url": f"https://source3.com/{topic}", "content": "..."},
    ]

async def generate_report(sources: list[dict]) -> str:
    await asyncio.sleep(2)  # giả lập write
    return f"Report based on {len(sources)} sources."


@mcp.tool(
    name="research",
    description="Research a given topic and produce a summary report"
)
async def research(
    topic: str = Field(description="Topic to research"),
    *,
    ctx: Context,
) -> str:
    # Notification 1: log ngay khi bắt đầu
    await ctx.info(f"Starting research on '{topic}'")

    # Notification 2: progress 20% — sắp fetch
    await ctx.report_progress(20, 100, "Fetching sources...")

    # Làm việc thật
    sources = await do_research(topic)

    # Notification 3: log milestone
    await ctx.info(f"Fetched {len(sources)} sources")

    # Notification 4: progress 70% — sắp synthesize
    await ctx.report_progress(70, 100, "Writing report...")

    # Làm việc thật
    results = await generate_report(sources)

    # Notification 5: progress 100%
    await ctx.report_progress(100, 100, "Done")
    await ctx.info("Report ready to deliver")

    return results


if __name__ == "__main__":
    mcp.run()

Dissect từng phần

1. Import Context

from mcp.server.fastmcp import FastMCP, Context

Context là class SDK cung cấp để tool "nói chuyện ngược" với client.

2. Tool signature với *, ctx: Context

async def research(topic: str = Field(...), *, ctx: Context,) -> str:

• * tách positional/keyword args. Claude gọi tool chỉ biết tham số topic.
• ctx keyword-only, không appear trong tool schema gửi cho Claude. SDK tự inject lúc runtime.

3. 5 notification được gửi

4. Không cần handle error

Nếu do_research() raise exception, SDK tự convert thành CallToolResult với isError=true. Không cần try/except cho notification flow.

client.py — Nhận và render notifications

File này minh họa một CLI client tối giản:

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


async def logging_callback(params: LoggingMessageNotificationParams):
    level = params.level.upper()
    print(f"[{level}] {params.data}")


async def print_progress_callback(
    progress: float,
    total: float | None,
    message: str | None,
):
    if total is not None:
        percentage = (progress / total) * 100
        print(f"Progress: {progress}/{total} ({percentage:.1f}%) — {message or ''}")
    else:
        print(f"Progress: {progress} — {message or ''}")


async def run():
    # Khai báo cách spawn server
    server_params = StdioServerParameters(
        command="uv",
        args=["run", "server.py"],
    )

    # Kết nối stdio
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read,
            write,
            logging_callback=logging_callback,  # gắn callback cho session
        ) as session:
            # Handshake bắt buộc
            await session.initialize()

            # Gọi tool với progress_callback riêng
            result = await session.call_tool(
                name="research",
                arguments={"topic": "AI in healthcare"},
                progress_callback=print_progress_callback,
            )

            # In kết quả cuối cùng
            print("\n=== FINAL RESULT ===")
            print(result.content[0].text if result.content else result)


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

Dissect các điểm quan trọng

1. Hai loại callback có scope khác nhau

ClientSession(
    read, write,
    logging_callback=logging_callback,  # ← session-level
)

session.call_tool(
    name="research",
    progress_callback=print_progress_callback,  # ← per-call
)

Log callback dùng cho MỌI tool call trong session. Progress callback chỉ dùng cho tool call cụ thể này — cho phép bạn custom render cho từng tool khác nhau.

2. stdio_client spawn subprocess

stdio_client(server_params) thực sự chạy uv run server.py dưới background và kết nối stdin/stdout. Đây là stdio transport từ bài 10.2.

3. Nhớ await session.initialize()

Bước này gửi handshake 3 message. Nếu bỏ, call_tool sẽ fail.

Chạy thực tế

cd notifications/
uv run client.py

Output sẽ giống:

[INFO] Starting research on 'AI in healthcare'
Progress: 20.0/100.0 (20.0%) — Fetching sources...
[INFO] Fetched 3 sources
Progress: 70.0/100.0 (70.0%) — Writing report...
Progress: 100.0/100.0 (100.0%) — Done
[INFO] Report ready to deliver

=== FINAL RESULT ===
Report based on 3 sources.

Flow timing:

• [INFO] Starting... hiện ngay lập tức
• Progress: 20% ngay sau
• ~2 giây silence (do_research sleep)
• [INFO] Fetched 3 sources + Progress: 70%
• ~2 giây silence (generate_report sleep)
• Progress: 100% + [INFO] Ready
• Kết quả cuối

Đây là cách user thấy một tool "chạy 4 giây" thay vì "đứng im 4 giây rồi hiện kết quả".

Modify project — Bài tập thực hành

Task 1: Đổi progress sang total=None (indeterminate)

Ở tool research, đổi 2 dòng progress:

await ctx.report_progress(progress=1, total=None, message="Fetching sources...")
await ctx.report_progress(progress=2, total=None, message="Writing report...")

Chạy lại. Output sẽ là:

Progress: 1 — Fetching sources...
Progress: 2 — Writing report...

(Không còn %.) Dùng khi server không biết total chính xác.

Task 2: Thêm error log

Modify do_research để giả lập 1 source bị skip:

async def do_research(topic: str, ctx: Context) -> list[dict]:
    sources = []
    for i in range(5):
        await asyncio.sleep(0.5)
        if i == 2:
            await ctx.warning(f"Source {i} returned 404, skipping")
            continue
        sources.append({"url": f"https://source{i}.com/{topic}", "content": "..."})
    return sources

Nhớ truyền ctx vào function. Run lại, observe warning line có hiện màu khác không (tùy terminal).

Task 3: Granular progress

Thay vì 3 giai đoạn cố định, progress theo số source đã fetch:

async def do_research(topic: str, ctx: Context) -> list[dict]:
    total = 20
    sources = []
    for i in range(total):
        await asyncio.sleep(0.1)
        sources.append({"url": f"https://s{i}.com/{topic}"})
        await ctx.report_progress(
            progress=len(sources),
            total=total,
            message=f"Fetched {len(sources)}/{total} sources"
        )
    return sources

Run lại, xem progress bar "mịn" hơn. Thử nghiệm: đổi asyncio.sleep(0.1) thành 0.01 — xem có bị flood log không, cảm nhận độ dày cadence.

Debug tips khi walkthrough không chạy

Lỗi 1: ModuleNotFoundError: mcp

Cause: Chưa chạy uv sync. Fix: cd notifications/ && uv sync.

Lỗi 2: Server spawn nhưng client không nhận notification

Cause: Có thể tool return quá nhanh, notification bị "flushed" trước khi callback chạy. Fix: Thêm await asyncio.sleep(0.01) giữa các notification để tạm tách.

Lỗi 3: TypeError: unexpected argument 'ctx'

Cause: SDK version cũ. Cần version có support inject Context. Fix: Update mcp package: uv add mcp --upgrade.

Lỗi 4: Progress hiển thị sai tỷ lệ

Cause: Có thể gửi progress=2 nhưng total=100 — UI render 2%. Fix: Đảm bảo scale consistent. Prefer progress=20, total=100 hoặc progress=0.2, total=1.0.

Observation — Đọc raw message bằng Inspector

Chạy project với Inspector thay vì client.py:

npx @modelcontextprotocol/inspector uv run server.py

Call tool research, quan sát message panel. Bạn sẽ thấy chính xác các JSON:

{"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","data":"Starting research on 'AI in healthcare'"}}

{"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"...","progress":20,"total":100,"message":"Fetching sources..."}}

{"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","data":"Fetched 3 sources"}}

...

{"jsonrpc":"2.0","id":"<tool-call-id>","result":{"content":[{"type":"text","text":"Report based on 3 sources."}]}}

Nhìn progressToken trong progress notification — đó là token match progress với tool call. Nếu client gọi 2 tool song song, mỗi tool có progressToken riêng, client route về đúng callback nhờ token này.

Tích hợp với Claude Desktop (client thật)

Thay vì dùng client.py demo, gắn server này vào Claude Desktop thực tế:

1. Config claude_desktop_config.json:

{
  "mcpServers": {
    "notifications-demo": {
      "command": "uv",
      "args": ["run", "/absolute/path/to/notifications/server.py"]
    }
  }
}

2. Restart Claude Desktop.

3. Trong Claude, hỏi

"Dùng tool research để research về quantum computing."

4. Quan sát

Claude Desktop hiển thị progress bar + log inline khi tool chạy.

Đây là trải nghiệm thực tế user sẽ có — chứ không phải CLI demo.

Anti-patterns từ walkthrough

Sao chép nguyên await asyncio.sleep vào production

Hiện tượng

Tool thật gọi API bên ngoài nhưng code demo dùng sleep() để fake.

Cách đúng

Dùng aiohttp hoặc httpx async cho API call thật. Notification pattern vẫn y chang.

Quên *, trong tool signature

Hiện tượng

async def tool(topic, ctx) → Claude thấy ctx trong schema, gọi với garbage.

Cách đúng

Luôn có *, ngay trước ctx: Context.

Gọi ctx.info() mà không await

Hiện tượng

ctx.info("hi") → warning "coroutine never awaited".

Cách đúng

Luôn await ctx.info(...), await ctx.report_progress(...).

Đưa code demo lên production không thay thế sleep

Hiện tượng

Tool production chạy nhanh xong vẫn sleep(2).

Cách đúng

Remove mọi sleep() sau khi thay bằng logic thật.

Áp dụng ngay

Bài tập 1: Research tool riêng của bạn (30 phút)

Bước 1

Clone project notifications/.

Bước 2

Thay research bằng tool thật của bạn. Ví dụ:

• query_database(sql) — progress theo số row.
• generate_content(prompt) — log theo step (draft, review, format).
• analyze_image(path) — progress theo region/object detected.

Bước 3

Deploy lên Claude Desktop, sử dụng 2-3 ngày.

Bước 4

Ghi lại:

• Có ít nhất 1 moment bạn "cứu" được user vì notification không? ___________
• Cadence có cần adjust không? ___________
• Có log level nào dùng chưa đúng không? ___________

Bài tập 2 (optional): Build web dashboard

Viết FastAPI app giữ session MCP → WebSocket push notifications ra browser → React component hiển thị progress bar. Đây là mini-project cho 1 buổi tối nhưng cực kỳ value.

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

Tài liệu tham khảo

• notifications.zip — project mẫu chính thức (lấy link download từ trang course gốc Anthropic Academy, hoặc clone từ python-sdk/examples)
• MCP Python SDK Examples — nhiều pattern khác
• Rich Progress Display — upgrade CLI UI