7.8 · Truy cập Resources từ client

Mở đầu: Biến "@mention" thành tính năng kỳ diệu

Flow chi tiết:

USER gõ:                 "Tóm tắt @report.pdf giúp tôi"
                                    │
                                    ▼
APP detect @mention:      mentions = ["report.pdf"]
                                    │
                                    ▼
APP gọi client method:    content = await mcp.read_resource(
                             "docs://documents/report.pdf"
                          )
                                    │
                                    ▼
CLIENT gửi request:       ReadResourceRequest { uri: ... }
                                    │
                                    ▼
SERVER return:            ReadResourceResult {
                            contents: [
                              { mimeType: "text/plain",
                                text: "The report details..." }
                            ]
                          }
                                    │
                                    ▼
CLIENT parse theo mime:   return resource.text (raw string)
                                    │
                                    ▼
APP inject vào prompt:    "<document name='report.pdf'>
                            The report details...
                           </document>

                           Tóm tắt @report.pdf giúp tôi"
                                    │
                                    ▼
CLAUDE nhận full context, trả lời ngay không cần tool call

Đây là pattern "context injection" — app quyết context gì Claude cần, fetch trước, inject thẳng. Pattern này tạo ra những trải nghiệm chat AI mượt mà hơn hẳn so với để Claude tự quyết tool call mỗi lần.

Extend MCPClient với resource methods

Từ Bài 7.6, MCPClient đã có list_tools() và call_tool(). Giờ thêm resource methods.

Import cần thiết

import json
from typing import Any
from pydantic import AnyUrl
from mcp import types

• json — parse JSON response
• AnyUrl từ Pydantic — validate URI format
• types — MCP type definitions

Method list_resources()

async def list_resources(self) -> list[types.Resource]:
    """List all Direct Resources from the server."""
    result = await self.session().list_resources()
    return result.resources


async def list_resource_templates(self) -> list[types.ResourceTemplate]:
    """List all Templated Resources (URIs with params)."""
    result = await self.session().list_resource_templates()
    return result.resourceTemplates

Pattern giống list_tools(). SDK return ListResourcesResult wrap list bên trong .resources.

Method read_resource()

async def read_resource(self, uri: str) -> Any:
    """Read a resource by URI. Auto-parses JSON if mime type indicates."""
    result = await self.session().read_resource(AnyUrl(uri))

    if not result.contents:
        return None

    resource = result.contents[0]

    if isinstance(resource, types.TextResourceContents):
        if resource.mimeType == "application/json":
            return json.loads(resource.text)
        return resource.text

    # Binary resource
    return resource.blob

Bóc tách chi tiết

AnyUrl(uri)

SDK yêu cầu AnyUrl object, không chấp nhận plain string. Pydantic validate URI format trước khi gửi.

result.contents[0]

contents là list. Tại sao không chỉ 1 content?

MCP spec cho phép một URI trả về nhiều content — ví dụ response đa phần (text + thumbnail). Thực tế >95% use case trả 1 content, nên chúng ta access [0].

isinstance(..., types.TextResourceContents)

Content có 2 type chính: TextResourceContents (text + mime) hoặc BlobResourceContents (binary). Check kiểu trước khi access field.

MIME-based branching

if resource.mimeType == "application/json":
    return json.loads(resource.text)

Nếu server nói là JSON → client tự parse. App không phải lo.

Nếu text/plain / text/markdown → return raw string.

Binary fallback

return resource.blob

Cho images, PDFs, etc. SDK tự base64-decode trước khi pass .blob.

Code đầy đủ

async def list_resources(self) -> list[types.Resource]:
    result = await self.session().list_resources()
    return result.resources

async def list_resource_templates(self) -> list[types.ResourceTemplate]:
    result = await self.session().list_resource_templates()
    return result.resourceTemplates

async def read_resource(self, uri: str) -> Any:
    result = await self.session().read_resource(AnyUrl(uri))
    resource = result.contents[0]
    if isinstance(resource, types.TextResourceContents):
        if resource.mimeType == "application/json":
            return json.loads(resource.text)
        return resource.text
    return resource.blob

Thêm vào mcp_client.py sau các tool methods.

Test client với resources

Quick test trong mcp_client.py:

async def main():
    async with MCPClient(
        command="uv",
        args=["run", "mcp_server.py"],
    ) as client:
        # List resources
        direct = await client.list_resources()
        templates = await client.list_resource_templates()
        print(f"Direct: {[r.uri for r in direct]}")
        print(f"Templates: {[r.uriTemplate for r in templates]}")

        # Read direct resource
        doc_list = await client.read_resource("docs://documents")
        print(f"All docs: {doc_list}")

        # Read templated resource
        content = await client.read_resource("docs://documents/plan.md")
        print(f"plan.md: {content}")


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

Chạy:

uv run mcp_client.py

Output:

Direct: ['docs://documents']
Templates: ['docs://documents/{doc_id}']
All docs: ['deposition.md', 'report.pdf', 'financials.docx', ...]
plan.md: The plan outlines the steps for the project's implementation.

Client giờ đã "biết đọc" resource. Bước tiếp là ghép vào chat UI.

Implement @mention trong CLI chat

Plan

1. Khi user gõ input, detect @word pattern
2. Với mỗi mention, fetch resource tương ứng
3. Wrap content với tag <document name='...'>...</document>
4. Inject vào prompt trước user query original

Code: CLI chat improvements

Update main.py:

import re

def extract_mentions(text: str) -> list[str]:
    """Extract @mentions like '@doc_id' or '@file-name.ext' from text."""
    return re.findall(r'@([\w\-.]+)', text)


async def build_augmented_prompt(
    user_input: str, mcp: MCPClient
) -> str:
    mentions = extract_mentions(user_input)
    if not mentions:
        return user_input

    context_parts = []
    for doc_id in mentions:
        try:
            content = await mcp.read_resource(f"docs://documents/{doc_id}")
            context_parts.append(
                f"<document name='{doc_id}'>\n{content}\n</document>"
            )
        except Exception as e:
            context_parts.append(
                f"<document name='{doc_id}' error='{e}' />"
            )

    return "\n\n".join(context_parts) + "\n\n" + user_input


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

    augmented = await build_augmented_prompt(raw_input, mcp)
    messages.append({"role": "user", "content": augmented})

    # ... rest of agentic loop

Flow test

User gõ: Tóm tắt @deposition.md trong 2 câu

1. extract_mentions → ["deposition.md"]
2. build_augmented_prompt gọi read_resource("docs://documents/deposition.md")
3. Server trả content
4. Prompt thực gửi Claude:

   <document name='deposition.md'>
   This deposition covers the testimony of Angela Smith, P.E.
   </document>

   Tóm tắt @deposition.md trong 2 câu

1. Claude tóm tắt ngay — không cần gọi tool.

UX advanced: Autocomplete

Professional implementation thêm:

if raw_input.endswith('@') or '@' in raw_input:
    available = await mcp.list_resource_templates()
    # Or if direct resource docs://documents exists:
    doc_list = await mcp.read_resource("docs://documents")
    # Show UI autocomplete với doc_list

Claude Desktop làm đúng pattern này — user thấy popup chọn document.

ReadResourceResult — Hiểu cấu trúc đầy đủ

Spec MCP define:

class ReadResourceResult:
    contents: list[ResourceContents]  # always list, thường 1 item

class TextResourceContents:
    uri: str              # URI gốc
    mimeType: str         # e.g. "text/plain"
    text: str             # nội dung text

class BlobResourceContents:
    uri: str
    mimeType: str
    blob: str             # base64-encoded binary

Khi nào contents có nhiều item?

Hiếm, nhưng spec cho phép cases:

• Multi-part response (text summary + binary preview)
• Collection nhỏ return cùng lúc
• Custom extension của server

Client của bạn có thể xử lý nâng cao:

async def read_resource_full(self, uri: str) -> list[Any]:
    """Return all content parts, not just first."""
    result = await self.session().read_resource(AnyUrl(uri))
    return [
        json.loads(c.text) if c.mimeType == "application/json"
        else c.text if hasattr(c, 'text')
        else c.blob
        for c in result.contents
    ]

Nhưng cho khóa học intro, contents[0] là đủ 99% case.

Bảng so sánh: Context injection patterns

Resource injection balance tốt nhất cho chat app có @-style feature: 1 HTTP call + token cost once + user explicit.

Ví dụ thực chiến: Chat app với autocomplete mention

Tình huống

Build CLI chat app polished: user gõ @, thấy autocomplete, chọn bằng Tab, content inject tự động.

Bước 1: Pre-fetch resource list (~1 phút)

Ở startup:

all_docs = await mcp.read_resource("docs://documents")

Bước 2: Detect @ trong input real-time

Dùng prompt_toolkit để có autocomplete UI:

from prompt_toolkit import PromptSession
from prompt_toolkit.completion import WordCompleter

completer = WordCompleter([f"@{doc}" for doc in all_docs])
session = PromptSession()

user_input = await session.prompt_async("Bạn > ", completer=completer)

User gõ @ → popup list doc names.

Bước 3: Fetch resources cho mentions

augmented = await build_augmented_prompt(user_input, mcp)

Bước 4: Send và display

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

Kết quả

User experience gần như Claude Desktop chính thống:

• Gõ @ → list docs
• Chọn, content auto-inject
• Claude respond ngay, no tool call

Thời gian build: ~2-3 giờ cho version polished.

Ví dụ theo ngành — Resource fetch patterns

Code editor với @file context

Editor

Cursor / Claude Code

Resource

file:///absolute/path/to/file.py

Pattern:

• User @mention file trong chat
• Editor fetch file qua MCP
• Inject content vào prompt
• Claude reason về code với full context

Dashboard app với @report

App

Internal BI tool

Resource

reports://weekly/{report_id}

Pattern:

• User @reference weekly report
• App fetch pre-computed metrics
• Claude generate narrative summary

CRM chat với @account

App

Sales enablement bot

Resource

crm://account/{domain}

Pattern:

• Rep @mention customer domain
• App fetch full account context (deals, interactions, contacts)
• Claude prep talking points instantly

Knowledge base assistant

App

Internal Q&A bot

Resources

kb://article/{slug}, kb://section/{id}

Pattern:

• User @mention article slug
• Bot fetch article + related sections
• Inject all → Claude answer with citations

Design review bot

App

Design QA assistant

Resource

figma://file/{key}/node/{id}

Pattern:

• Designer @mention Figma component
• App fetch component metadata + image export
• Claude review design vs guidelines

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

Fetch resource đồng bộ một cách sequential

Sai lầm:

for mention in mentions:
    content = await mcp.read_resource(f"docs://documents/{mention}")
    parts.append(content)

Tại sao là sai

10 mentions = 10 round-trip. Latency tuyến tính.

Cách đúng

Parallel với asyncio.gather:

contents = await asyncio.gather(*[
    mcp.read_resource(f"docs://documents/{m}")
    for m in mentions
])

Không cache resource đã fetch

Sai lầm

User hỏi 5 câu về cùng @report.pdf. App fetch 5 lần.

Tại sao là sai

Waste bandwidth, waste time.

Cách đúng

Cache theo URI với TTL:

from functools import lru_cache

Fetch hết mọi resource cùng lúc startup

Sai lầm:

for uri in all_uris:
    content = await mcp.read_resource(uri)
    cache[uri] = content

Tại sao là sai

Slow startup, memory bloat, data có thể không dùng.

Cách đúng

Lazy fetch khi cần. Cache sau first access.

Leak error details lên user

Sai lầm:

try:
    content = await mcp.read_resource(uri)
except Exception as e:
    print(f"Error: {e}")  # có thể chứa stack trace, paths nhạy cảm

Tại sao là sai: Security leak. Production app show internal paths = bad.

Cách đúng:

except Exception as e:
    logger.exception("resource fetch failed")
    context_parts.append(f"<document name='{mention}' unavailable='true'/>")

Inject toàn bộ resource không check size

Sai lầm:

content = await mcp.read_resource(uri)  # 5MB text
prompt += content  # prompt bloat

Tại sao là sai

Vượt context limit, cost cao.

Cách đúng

Check size, truncate hoặc summarize trước:

if len(content) > 50_000:
    content = content[:50_000] + "\n[... truncated, use tools to see more]"

Trust mime_type blindly

Sai lầm:

if resource.mimeType == "application/json":
    data = json.loads(resource.text)  # crash nếu server nói JSON nhưng text invalid

Tại sao là sai: Malformed server breaks client.

Cách đúng:

try:
    data = json.loads(resource.text)
except json.JSONDecodeError:
    logger.warning("mime says JSON but invalid, returning raw")
    data = resource.text

Mẹo nâng cao

Mẹo 1: Prefetch với speculative loading

Khi user gõ @re, prefetch resource match autocomplete (report.pdf, request.md):

async def speculative_prefetch(prefix: str):
    candidates = [d for d in all_docs if d.startswith(prefix)]
    for doc in candidates[:3]:  # top 3
        asyncio.create_task(mcp.read_resource(f"docs://documents/{doc}"))

User feel latency gần-zero khi mention.

Mẹo 2: Resource subscription cho real-time update

await mcp.session().subscribe_resource(AnyUrl(uri))

@mcp.on_resource_updated
async def handler(uri, new_content):
    cache[uri] = new_content
    # UI update toast

Use case: collaborative editing, live data.

Mẹo 3: Template substitution server-side

Thay vì inject raw content, server trả pre-formatted prompt:

@mcp.resource("prompt://doc-summary/{doc_id}")
def doc_summary_prompt(doc_id: str) -> str:
    content = docs[doc_id]
    return f"Summarize this document in 3 bullets:\n\n{content}"

Client chỉ inject URL response → prompt already formatted.

Mẹo 4: Observability — track resource usage

@timed
@log_metric("resource_fetch")
async def read_resource(self, uri):
    ...

Measure: fetch latency, failure rate, content size distribution. Dùng cho tuning.

Áp dụng ngay

Bài tập 1: Extend MCPClient (~15 phút)

Bước 1

Mở mcp_client.py. Thêm imports: json, AnyUrl.

Bước 2

Implement 3 method: list_resources, list_resource_templates, read_resource.

Bước 3

Update main() test harness:

async def main():
    async with MCPClient(...) as client:
        resources = await client.list_resources()
        for r in resources:
            print(f"- {r.uri}")
        content = await client.read_resource("docs://documents")
        print(f"docs: {content}")

Bước 4: Chạy uv run mcp_client.py. Ghi lại:

• Số direct resources: ___________
• Số templates: ___________
• Output docs://documents có parse JSON không? ___________

Bài tập 2: @mention trong chat (~25 phút)

Bước 1

Mở main.py. Import re.

Bước 2

Viết hàm extract_mentions và build_augmented_prompt như bài.

Bước 3

Replace messages.append({"role": "user", "content": user_input}) bằng:

augmented = await build_augmented_prompt(user_input, mcp)
messages.append({"role": "user", "content": augmented})

Bước 4: Test:

• Câu 1: "Tóm tắt @plan.md trong 1 câu."
• Câu 2: "@plan.md và @spec.txt có gì khác nhau?"
• Câu 3: "@nonexistent.md là gì?" (test error handling)

Bước 5: Ghi lại:

• Claude có cần tool call nào không cho câu 1? ___________
• Câu 2 (2 mentions) có chạy đúng không? ___________
• Câu 3 app handle graceful không crash? ___________

Bài tập 3 (thử thách): Cache layer (~15 phút)

Thêm cache đơn giản với TTL 60 giây:

import time
from collections import OrderedDict

class MCPClient:
    def __init__(self, ...):
        ...
        self._resource_cache = OrderedDict()
        self._cache_ttl = 60  # seconds

    async def read_resource_cached(self, uri: str):
        now = time.time()
        if uri in self._resource_cache:
            content, ts = self._resource_cache[uri]
            if now - ts < self._cache_ttl:
                return content

        content = await self.read_resource(uri)
        self._resource_cache[uri] = (content, now)
        return content

Test:

• Call read_resource_cached 2 lần cùng URI trong 10s
• Lần 2 có nhanh hơn không? (ms difference)
• Wait 65s, call lần 3 → có invalidate cache?

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

Tài liệu tham khảo

• MCP spec — Resources
• Pydantic AnyUrl
• asyncio.gather
• "Accessing resources" — Anthropic Academy video