Truy cập Resources từ client

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

Method list_resources()

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

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

Method list_resources()

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

Lưu ý: 2 method riêng cho direct vs template. MCP spec tách chúng vì client UI thường hiển thị khác nhau (direct = button, template = form với input).

Method read_resource()

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

Method read_resource()

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

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

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

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

Extend MCPClient với resource methods (tiếp)

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

Code đầy đủ

return resource.blob

Code đầy đủ

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

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

Test client với resources

Quick test trong mcp_client.py:

Chạy:

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

Test client với resources (tiếp)

Output:

uv run mcp_client.py

Test client với resources (tiếp)

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

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.

Implement @mention trong CLI chat

Plan

Code: CLI chat improvements

Update main.py:

Flow test

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

• Khi user gõ input, detect @word pattern
• Với mỗi mention, fetch resource tương ứng
• Wrap content với tag <document name='...'>...</document>
• Inject vào prompt trước user query original
• extract_mentions → ["deposition.md"]
• build_augmented_prompt gọi read_resource("docs://documents/deposition.md")
• Server trả content
• Prompt thực gửi Claude:

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


# Trong chat_loop:
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

UX advanced: Autocomplete

Professional implementation thêm:

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

   <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

UX advanced: Autocomplete

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

# Khi user gõ '@', hiển thị autocomplete từ list_resources
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

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

Spec MCP define:

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

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

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

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

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?

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

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
    ]

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:

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

Dùng prompt_toolkit để có autocomplete UI:

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

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

User gõ @ → popup list doc names.

Bước 3: Fetch resources cho mentions

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)

Bước 3: Fetch resources cho mentions

Bước 4: Send và display

augmented = await build_augmented_prompt(user_input, mcp)

Bước 4: Send và display

Kết quả

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

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

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

messages.append({"role": "user", "content": augmented})
# ... rest of loop

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:

📊 Dashboard app với @report

App: Internal BI tool

Resource: reports://weekly/{report_id}

Pattern:

💼 CRM chat với @account

App: Sales enablement bot

Resource: crm://account/{domain}

Pattern:

📚 Knowledge base assistant

App: Internal Q&A bot

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

Pattern:

🎨 Design review bot

App: Design QA assistant

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

Pattern:

• User @mention file trong chat
• Editor fetch file qua MCP
• Inject content vào prompt
• Claude reason về code với full context
• User @reference weekly report
• App fetch pre-computed metrics
• Claude generate narrative summary
• Rep @mention customer domain
• App fetch full account context (deals, interactions, contacts)
• Claude prep talking points instantly
• User @mention article slug
• Bot fetch article + related sections
• Inject all → Claude answer with citations
• 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:

Tại sao là sai: 10 mentions = 10 round-trip. Latency tuyến tính.

Cách đúng: Parallel với asyncio.gather:

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

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

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

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

❌ Không cache resource đã fetch

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

Sai lầm:

from functools import lru_cache
# Hoặc TTL cache như cachetools

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

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:

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

❌ Leak error details lên user

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

Cách đúng:

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

Anti-patterns — Những sai lầm cần tránh (tiếp)

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

Sai lầm:

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

Tại sao là sai: Vượt context limit, cost cao.

Cách đúng: Check size, truncate hoặc summarize trước:

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

Anti-patterns — Những sai lầm cần tránh (tiếp)

❌ Trust mime_type blindly

Sai lầm:

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

❌ Trust mime_type blindly

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

Cách đúng:

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

Anti-patterns — Những sai lầm cần tránh (tiếp)

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

User feel latency gần-zero khi mention.

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

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}"))

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

Use case: collaborative editing, live data.

Mẹo 3: Template substitution server-side

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

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

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

Mẹo 3: Template substitution server-side

Client chỉ inject URL response → prompt already formatted.

Mẹo 4: Observability — track resource usage

# Server
@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}"

Mẹo 4: Observability — track resource usage

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

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

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

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

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:

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

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ài tập 2: @mention trong chat (~25 phút)

Bước 4: Test:

Bước 5: Ghi lại:

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:

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

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

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

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?

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

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

🎯 Client read_resource(uri) parse theo mime_type — JSON auto json.loads, text raw, binary .blob. Code app không phải lo.

🎯 result.contents[0] là pattern 99% case — Spec cho phép multi-content nhưng thực tế hiếm. Access first item.

🎯 @mention = detect + fetch + inject — Extract mentions bằng regex, fetch parallel với asyncio.gather, wrap <document> tag inject prompt.

🎯 Resource > Tool khi app biết chắc cần — Pre-fetch tiết kiệm 1 turn + token. Tool tốt cho Claude autonomous, resource tốt cho explicit user reference.

🎯 Cache + parallel + error handling là kỹ thuật production — Không ship resource fetch without 3 thứ này nếu traffic >10 req/phút.