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

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

Project mẫu có 5 file chính:

Download: link notifications.zip có trong bài 10.6 gốc. Hoặc clone từ repo Anthropic chính thứ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

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

Dissect từng phần

1. Import Context

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

# Tạo MCP server với 1 cái tên
mcp = FastMCP(name="notifications-demo")

# Giả lập function research lâu
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

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

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

from mcp.server.fastmcp import FastMCP, Context

server.py — Emit notifications (tiếp)

3. 5 notification được gửi

Lưu ý: không có progress 0%, vì progress 20% đã communicate "đang start".

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.

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

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

client.py — Nhận và render notifications

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

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

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

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


# Callback cho log — chạy khi server gửi notifications/message
async def logging_callback(params: LoggingMessageNotificationParams):
    level = params.level.upper()
    print(f"[{level}] {params.data}")


# Callback cho progress — chạy khi server gửi notifications/progress
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

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.

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

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

Chạy thực tế

Output sẽ giống:

cd notifications/
uv run client.py

Chạy thực tế (tiếp)

Flow timing:

Đâ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ả".

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

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

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:

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

await ctx.report_progress(progress=1, total=None, message="Fetching sources...")
# ... làm việc ...
await ctx.report_progress(progress=2, total=None, message="Writing report...")

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

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

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

Task 2: Thêm error log

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

Task 3: Granular progress

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.

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

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:

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

npx @modelcontextprotocol/inspector uv run server.py

Observation — Đọc raw message bằng Inspector (tiếp)

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.

{"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."}]}}

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:

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.

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

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

Bước 3: Deploy lên Claude Desktop, sử dụng 2-3 ngày.

Bước 4: Ghi lại:

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.

• query_database(sql) — progress theo số row.
• generate_content(prompt) — log theo step (draft, review, format).
• analyze_image(path) — progress theo region/object detected.
• 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? ___________

Tóm tắt walkthrough

🎯 Project mẫu notifications có 5 file — tối thiểu đủ để hiểu + modify + chạy thực tế.

🎯 Context injection tự động — ctx: Context sau *, trong tool signature. SDK tự inject.

🎯 5 notification trong research tool — log start + milestone + end, progress 20/70/100%.

🎯 Client có 2 callback — logging_callback session-level, progress_callback per-call.

🎯 progressToken match progress với tool — khi nhiều tool song song, token route đúng callback.

🎯 Chuyển sang tool thật cần thay sleep bằng async IO — pattern notifications giữ nguyên, logic bên trong đổi.