Walkthrough — Triển khai roots với project thật

Project overview — Architecture phức tạp hơn

Khác với 2 walkthrough trước (5 file phẳng), project roots/ có cấu trúc module:

Lý do architecture này: project này là full chat client với Claude, không chỉ demo MCP. main.py spawn client → client connect tới MCP server → tool convert_video có logic thật trong video_converter.py.

Setup:

roots/
├── .env.example
├── .gitignore
├── main.py                     # Entry point — spawn client
├── mcp_client.py               # MCPClient wrapper
├── mcp_server.py               # MCP server
├── pyproject.toml
├── README.md
└── core/                       # Business logic tách module
    ├── __init__.py
    ├── chat.py                 # Chat loop
    ├── claude.py               # Anthropic SDK wrapper
    ├── cli_chat.py             # CLI interface
    ├── cli.py                  # CLI args parser
    ├── tools.py                # Tool orchestration
    ├── utils.py                # Helper functions
    └── video_converter.py      # Business logic: convert video

cd ~/mcp-advanced
unzip roots.zip
cd roots/

cp .env.example .env
# Edit .env, add ANTHROPIC_API_KEY

uv sync

# ffmpeg required cho video conversion
brew install ffmpeg  # macOS
# hoặc: sudo apt install ffmpeg  # Ubuntu

main.py — Entry point & CLI args pattern

Đoạn quan trọng nhất:

Key observations

1. CLI args = root paths

# main.py (key excerpt)
import sys
import asyncio
from contextlib import AsyncExitStack
from core.chat import Chat
from core.claude import Claude
from mcp_client import MCPClient


async def main():
    claude_model = "claude-sonnet-5"
    claude = Claude(model=claude_model)

    # Lấy root paths từ CLI arguments
    root_paths = sys.argv[1:]
    if not root_paths:
        print("Usage: uv run main.py <root1> [root2] ...")
        print("Example: uv run main.py /path/to/videos /another/path")
        sys.exit(1)

    clients = {}
    async with AsyncExitStack() as stack:
        # Create the MCP client với root directories
        doc_client = await stack.enter_async_context(
            MCPClient(
                command="uv",
                args=["run", "mcp_server.py"],
                roots=root_paths,  # ← inject roots vào client
            )
        )
        clients["doc_client"] = doc_client

        # Start chat loop
        chat = Chat(clients=clients, claude_service=claude)
        await chat.start()


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

Key observations

Mọi argument sau main.py trở thành path. Đơn giản, explicit, dễ audit.

2. Defensive check

uv run main.py /Users/jimmy/Movies /Users/jimmy/Desktop

main.py — Entry point & CLI args pattern (tiếp)

Server cần ít nhất 1 root để hoạt động. Không có root = không có gì để access.

3. AsyncExitStack cho cleanup

if not root_paths:
    print("Usage: ...")
    sys.exit(1)

main.py — Entry point & CLI args pattern (tiếp)

Pattern này đảm bảo mọi context manager cleanup khi stack exit. Production-grade.

async with AsyncExitStack() as stack:
    doc_client = await stack.enter_async_context(MCPClient(...))

mcp_client.py — MCPClient wrapper

Đây là lớp wrapper quan trọng — nó handle list_roots_callback transparently cho developer.

Dissect

1. list_roots_callback là async method của class

Mỗi khi server emit roots/list request, SDK gọi callback này. Callback return ListRootsResult.

2. file:// URI scheme

# mcp_client.py (simplified key parts)
from pathlib import Path
from contextlib import asynccontextmanager

from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp import ClientSession
from mcp.types import Root, ListRootsResult


class MCPClient:
    def __init__(self, command: str, args: list[str], roots: list[str]):
        self.command = command
        self.args = args
        self.roots = roots  # ← store roots
        self.session: ClientSession | None = None

    async def list_roots_callback(self) -> ListRootsResult:
        """SDK gọi callback này khi server request roots."""
        root_objects = [
            Root(
                uri=f"file://{Path(p).resolve()}",
                name=Path(p).name or p,
            )
            for p in self.roots
        ]
        return ListRootsResult(roots=root_objects)

    async def __aenter__(self):
        server_params = StdioServerParameters(
            command=self.command,
            args=self.args,
        )
        self._stdio_ctx = stdio_client(server_params)
        read, write = await self._stdio_ctx.__aenter__()

        self._session_ctx = ClientSession(
            read, write,
            list_roots_callback=self.list_roots_callback,  # ← register
        )
        self.session = await self._session_ctx.__aenter__()
        await self.session.initialize()
        return self

    async def __aexit__(self, *args):
        await self._session_ctx.__aexit__(*args)
        await self._stdio_ctx.__aexit__(*args)

    async def call_tool(self, name: str, arguments: dict):
        return await self.session.call_tool(name=name, arguments=arguments)

Dissect

resolve() đưa về absolute path. URI chuẩn format file:///absolute/path.

3. Re-use pattern

MCPClient là abstraction bạn có thể sao chép cho project khác. Thay list_roots_callback bằng logic của bạn (đọc config file, query DB, ...).

Root(uri=f"file://{Path(p).resolve()}", name=...)

mcp_server.py — Server side với enforcement

Dissect — Best practices shown

1. Check cả input và output path

# mcp_server.py
from mcp.server.fastmcp import FastMCP, Context
from pathlib import Path
from pydantic import Field

from core.video_converter import convert_video as do_convert_video

mcp = FastMCP(name="video-tools")


import sys
from urllib.parse import urlparse, unquote


def _uri_to_path(uri: str) -> Path:
    """file:// URI → Path, handling Windows drive letters."""
    parsed = urlparse(uri)
    path = unquote(parsed.path)
    if sys.platform == "win32" and path.startswith("/") and len(path) > 2 and path[2] == ":":
        path = path[1:]  # '/C:/foo' → 'C:/foo'
    return Path(path).resolve()


async def is_path_allowed(ctx: Context, requested_path: str) -> bool:
    """Check if path is inside any approved root."""
    roots_result = await ctx.session.list_roots()
    requested = Path(requested_path).resolve()

    for root in roots_result.roots:
        root_path = _uri_to_path(root.uri)
        try:
            requested.relative_to(root_path)
            return True
        except ValueError:
            continue

    return False


@mcp.tool(
    name="list_files",
    description="List files in an accessible directory"
)
async def list_files(
    directory: str = Field(description="Directory path to list"),
    *,
    ctx: Context,
) -> str:
    if not await is_path_allowed(ctx, directory):
        return f"[ERROR] {directory} is outside approved roots."

    path = Path(directory)
    if not path.is_dir():
        return f"[ERROR] {directory} is not a directory."

    files = [f.name for f in path.iterdir()]
    return f"Files in {directory}:\n" + "\n".join(files)


@mcp.tool(
    name="convert_video",
    description="Convert video between formats (MP4, MOV, AVI, etc.)"
)
async def convert_video(
    input_path: str = Field(description="Path to input video"),
    output_format: str = Field(description="Output format: mp4, mov, avi, mkv"),
    *,
    ctx: Context,
) -> str:
    # 1. Security check input
    if not await is_path_allowed(ctx, input_path):
        return f"[ERROR] Input {input_path} is outside approved roots."

    # 2. Build output path
    in_path = Path(input_path).resolve()
    out_path = in_path.with_suffix(f".{output_format}")

    # 3. Security check output (should also be in roots)
    if not await is_path_allowed(ctx, str(out_path)):
        return f"[ERROR] Output {out_path} would be outside approved roots."

    # 4. Notify user
    await ctx.info(f"Converting {input_path} → {out_path}")
    await ctx.report_progress(0, 100, "Starting conversion...")

    try:
        # 5. Do actual conversion (notifications inside)
        await do_convert_video(str(in_path), str(out_path), ctx)
    except Exception as e:
        return f"[ERROR] Conversion failed: {e}"

    await ctx.report_progress(100, 100, "Done")
    return f"Converted: {out_path}"


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

Dissect — Best practices shown

User có thể nói "convert biking.mp4 to /etc/malicious.mov" → output ngoài roots → reject.

2. Graceful error message

Thay vì raise exception, return error message rõ. Claude có thể relay cho user hiểu rõ vì sao fail.

3. Notifications inline

if not await is_path_allowed(ctx, input_path): ...
if not await is_path_allowed(ctx, str(out_path)): ...

mcp_server.py — Server side với enforcement (tiếp)

Long-running tool → user see progress. Kết hợp bài 10.6.

4. Business logic tách module

await ctx.info(f"Converting {input_path} → {out_path}")
await ctx.report_progress(0, 100, "Starting conversion...")

mcp_server.py — Server side với enforcement (tiếp)

Tool function mỏng — security + orchestration. Heavy work ở video_converter.py. Dễ test, dễ reuse.

from core.video_converter import convert_video as do_convert_video

core/video_converter.py — Business logic

Dissect

1. Run ffmpeg as subprocess

# core/video_converter.py (simplified)
import asyncio
import subprocess
from pathlib import Path

from mcp.server.fastmcp import Context


async def convert_video(input_path: str, output_path: str, ctx: Context):
    """Invoke ffmpeg to convert. Emit progress via ctx."""
    cmd = [
        "ffmpeg",
        "-i", input_path,
        "-y",  # overwrite output
        output_path,
    ]

    await ctx.info(f"Running: {' '.join(cmd)}")

    # Spawn ffmpeg, stream stderr (ffmpeg prints progress ra stderr)
    process = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    # Monitor progress (parse ffmpeg output)
    progress = 10
    while True:
        line = await process.stderr.readline()
        if not line:
            break
        line_str = line.decode(errors="replace").strip()

        # Progress simulation
        if "frame=" in line_str:
            progress = min(progress + 5, 90)
            await ctx.report_progress(progress, 100, line_str[:60])

    await process.wait()

    if process.returncode != 0:
        stderr_full = await process.stderr.read()
        raise RuntimeError(f"ffmpeg failed: {stderr_full.decode()}")

Dissect

Async so main tool function không block.

2. Parse stderr cho progress

FFmpeg log progress ra stderr. Parse line để emit MCP progress notification.

3. Error handle return code ≠ 0

Raise RuntimeError → tool catch → report error message.

asyncio.create_subprocess_exec(*cmd, stdout=..., stderr=...)

Chạy project

Chat interface xuất hiện. Thử:

cd roots/
uv run main.py ~/Movies ~/Desktop

Chạy project (tiếp)

Wait — bị reject. Vì Claude gửi path relative, không match root. Fix bằng cách:

Solution A: Prompt engineer để Claude dùng full path:

You: convert biking.mp4 to mov
Claude: Let me convert that for you...
[Tool call: convert_video(input_path="biking.mp4", output_format="mov")]
[ERROR] biking.mp4 is outside approved roots.

Chạy project (tiếp)

Solution B: Tool auto-search trong roots:

# In system prompt:
"When users reference files, use full absolute paths based on available roots."

Chạy project (tiếp)

Thử lại:

@mcp.tool()
async def convert_video(input_path, output_format, *, ctx):
    # If not absolute, search in roots
    if not Path(input_path).is_absolute():
        roots = await ctx.session.list_roots()
        for root in roots.roots:
            candidate = Path(root.uri.replace("file://", "")) / input_path
            if candidate.exists():
                input_path = str(candidate)
                break
        else:
            return f"[ERROR] File {input_path} not found in approved roots."

    # ... rest

Chạy project (tiếp)

End-to-end thành công: Claude tự tìm file trong roots → convert → output cũng trong roots. Secure và ergonomic.

You: convert biking.mp4 to mov
Claude: [Tool call: convert_video(...)]
Converting /Users/jimmy/Movies/biking.mp4 → /Users/jimmy/Movies/biking.mov
[Progress 25% ...]
[Progress 50% ...]
[Progress 90% ...]
Done! File created at /Users/jimmy/Movies/biking.mov

Mở rộng project — Bài tập modify

Exercise 1: Add read_file tool

Note: size limit quan trọng — prevent reading /dev/zero hoặc massive log file.

Exercise 2: Add write_file tool với confirmation

@mcp.tool()
async def read_file(
    file_path: str,
    *,
    ctx: Context,
) -> str:
    if not await is_path_allowed(ctx, file_path):
        return f"[ERROR] {file_path} is outside approved roots."

    path = Path(file_path)
    if not path.is_file():
        return f"[ERROR] {file_path} is not a file."

    # Size limit
    if path.stat().st_size > 1_000_000:
        return f"[ERROR] File too large (>{1_000_000} bytes)"

    content = path.read_text(errors="replace")
    return content

Exercise 2: Add write_file tool với confirmation

Prod version nên có confirmation prompt qua sampling:

@mcp.tool()
async def write_file(
    file_path: str,
    content: str,
    *,
    ctx: Context,
) -> str:
    if not await is_path_allowed(ctx, file_path):
        return f"[ERROR] {file_path} is outside approved roots."

    await ctx.warning(f"About to write {len(content)} bytes to {file_path}")

    path = Path(file_path)
    path.write_text(content)

    return f"Wrote {len(content)} bytes to {file_path}"

Mở rộng project — Bài tập modify (tiếp)

Exercise 3: Test path traversal attacks

Create file roots/attack-test.sh:

# Ask user confirmation via sampling
confirm = await ctx.session.create_message(
    messages=[SamplingMessage(role="user", content=TextContent(
        type="text",
        text=f"User wants to write to {file_path}. Do you approve? (yes/no)"
    ))],
    max_tokens=10,
)
if "yes" not in confirm.content.text.lower():
    return "[CANCELLED] User did not approve."

Exercise 3: Test path traversal attacks

Create symlink: ln -s /etc/passwd /tmp/sandbox/symlink-to-etc-passwd.

Run. Expect: all reject (denied). Nếu có pass → bug trong is_path_allowed.

#!/bin/bash
# Test path traversal attempts.
# Run with: bash attack-test.sh

uv run main.py /tmp/sandbox <<EOF
read_file /etc/passwd
read_file ../../../etc/passwd
read_file /tmp/sandbox/../../../etc/passwd
read_file /tmp/sandbox/symlink-to-etc-passwd
EOF

Tích hợp với Claude Desktop

Gắn server này vào Claude Desktop:

Issue: args ở đây đi vào main.py, nhưng main.py lại spawn mcp_server.py — roots không được forward đúng chỗ. Cần refactor:

• Tách mcp_server.py đứng riêng (không cần main.py wrapper).
• Đọc roots từ env var thay vì CLI args.

{
  "mcpServers": {
    "video-tools": {
      "command": "uv",
      "args": ["run", "/path/to/roots/main.py", "/Users/jimmy/Movies"]
    }
  }
}

Tích hợp với Claude Desktop (tiếp)

Pattern này decouple client từ server invocation — cleaner cho real-world.

# mcp_server.py refactor
import os

ROOTS = os.environ.get("MCP_ROOTS", "").split(":")
# In Claude Desktop config:
# "env": { "MCP_ROOTS": "/Users/jimmy/Movies:/Users/jimmy/Desktop" }

Debug tips

Tip 1: Log roots khi initialize

Tip 2: Inspector shows roots dialog

MCP Inspector có panel "Roots" — dev config roots trực tiếp từ UI, không cần CLI args. Hữu ích cho quick test.

Tip 3: Test với empty roots

# Trong mcp_server.py, có thể dùng @mcp.startup() hook nếu SDK support
# Hoặc log trong tool đầu tiên được gọi
await ctx.info(f"Roots: {[r.name for r in (await ctx.session.list_roots()).roots]}")

Tip 3: Test với empty roots

Expected: error "Usage: ..." + exit. Handle gracefully trong production.

Tip 4: Permission denied từ OS

Dù bạn grant root, OS vẫn cần permission. File chmod 600 bởi user khác → server không read được dù trong root.

uv run main.py

Anti-patterns từ walkthrough

❌ Root path hardcode trong mcp_server.py

Hiện tượng: ROOTS = ["/Users/jimmy/Movies"] trong code.

Cách đúng: Nhận từ client dynamic qua list_roots(). Code reusable.

❌ Skip Path.resolve()

Hiện tượng: Compare raw path → vulnerable.

Cách đúng: Luôn .resolve() trước khi compare.

❌ Không sanitize output format

Hiện tượng: output_format từ user → thành extension file. User inject ../../evil → bypass.

Cách đúng:

❌ Unicode normalization missing

Hiện tượng: Path có ký tự Unicode → NFC/NFD form khác nhau → compare fail.

Cách đúng: unicodedata.normalize('NFC', path) trước compare trên macOS (macOS dùng NFD tiềm ẩn).

if not re.match(r'^[a-z0-9]+$', output_format):
    return "[ERROR] Invalid format"

Áp dụng ngay

Bài tập 1: Full security audit (30 phút)

Bước 1: Clone roots/ project.

Bước 2: Thử 10 attack path:

Bước 3: Tool nên reject ALL. Ghi ra case nào pass (= bug).

Bước 4: Fix bugs, test lại.

Bài tập 2 (optional): Build roots-aware tool của bạn

Build tool cho use case của bạn (PDF reader, git log viewer, ...) với:

Deploy vào Claude Desktop, dùng 1 tuần, note những pattern tốt/tệ.

• ../../../etc/passwd
• Symlink to outside
• Unicode variants
• Case variations
• Trailing slashes
• Double slashes //
• Whitespace in path
• NULL byte \x00
• Very long path (> 4096 chars)
• Empty string
• Roots enforcement.
• Notifications.
• Size limits.
• Audit log.

Tóm tắt walkthrough

🎯 Project roots/ phức tạp hơn — có module core/, CLI args, wrapper MCPClient.

🎯 Roots cung cấp qua CLI args — explicit và dễ audit.

🎯 MCPClient.list_roots_callback là entry point — SDK gọi khi server request.

🎯 is_path_allowed phải dùng Path.resolve().relative_to() — string match là bug.

🎯 Check cả input và output path — attacker có thể trick qua output.

🎯 Tool mỏng, business logic ở module riêng — ergonomic và testable.

🎯 OS permission vẫn là layer cuối — roots không bypass filesystem permission.