MCP Resource Exposure Patterns

MCP Resource Exposure Patterns Skill

Metadata (Tier 1)

Keywords: resources, uri, context provisioning, list_resources, read_resource

File Patterns: **/resources/*.py, **/providers.py

Modes: backend_python

Instructions (Tier 2)

Resources vs Tools

Resources (app-controlled):

• Provide read-only context to Claude
• URI-based identification
• Examples: documentation, logs, database schemas, API specs

Tools (model-controlled):

• Perform actions with side effects
• Require explicit invocation by Claude
• Examples: create_file, execute_query, send_email

Rule: If it's read-only context provisioning → Resource. If it performs an action → Tool.

URI Patterns

Resources use URI schemes for identification:

# File resources
"file:///project/README.md"
"file:///docs/api/authentication.md"

# Database resources
"db:///schemas/users"
"db:///tables/orders/schema"

# API resources
"api:///endpoints/v1/users"
"api:///swagger.json"

# Log resources
"log:///app/2024-01-15"
"log:///errors/recent"

# Custom schemes
"config:///settings/database"
"metric:///cpu/usage"

list_resources Implementation

from pydantic import BaseModel, ConfigDict
from typing import Literal

class ResourceDescriptor(BaseModel):
    """Resource metadata returned by list_resources."""
    model_config = ConfigDict(strict=True)

    uri: str
    name: str
    mimeType: str
    description: str | None = None

@server.list_resources()
async def list_resources() -> list[dict]:
    """List all available resources."""
    resources = []

    # File-based resources
    docs_dir = Path("docs")
    if docs_dir.exists():
        for md_file in docs_dir.rglob("*.md"):
            resources.append(
                ResourceDescriptor(
                    uri=f"file:///{md_file}",
                    name=md_file.stem,
                    mimeType="text/markdown",
                    description=f"Documentation: {md_file.stem}"
                ).model_dump()
            )

    # Database schema resources
    db_schemas = await get_database_schemas()
    for schema in db_schemas:
        resources.append(
            ResourceDescriptor(
                uri=f"db:///schemas/{schema.name}",
                name=f"{schema.name} Schema",
                mimeType="application/json",
                description=f"Database schema for {schema.name}"
            ).model_dump()
        )

    return resources

read_resource Implementation

import aiofiles
from pathlib import Path

class ResourceContent(BaseModel):
    """Content returned by read_resource."""
    model_config = ConfigDict(strict=True)

    uri: str
    text: str | None = None
    blob: str | None = None  # Base64 encoded binary
    mimeType: str

@server.read_resource()
async def read_resource(uri: str) -> dict:
    """Read resource content by URI."""

    # Validate URI scheme
    if uri.startswith("file:///"):
        return await read_file_resource(uri)
    elif uri.startswith("db:///"):
        return await read_database_resource(uri)
    elif uri.startswith("log:///"):
        return await read_log_resource(uri)
    else:
        raise ValueError(f"Unsupported URI scheme: {uri}")

async def read_file_resource(uri: str) -> dict:
    """Read file-based resource."""
    # Security: validate path
    path = uri.removeprefix("file://")
    path = Path(path).resolve()

    # Prevent path traversal
    if ".." in str(path):
        raise ValueError("Path traversal not allowed")

    if not path.exists():
        raise FileNotFoundError(f"Resource not found: {uri}")

    # Async file reading
    async with aiofiles.open(path, "r") as f:
        content = await f.read()

    return {
        "contents": [
            ResourceContent(
                uri=uri,
                text=content,
                mimeType=get_mime_type(path)
            ).model_dump()
        ]
    }

Dynamic Resources

@server.list_resources()
async def list_resources() -> list[dict]:
    """List resources with dynamic content."""

    # Current date logs (past 7 days)
    resources = []
    for days_ago in range(7):
        date = datetime.now() - timedelta(days=days_ago)
        date_str = date.strftime("%Y-%m-%d")

        resources.append(
            ResourceDescriptor(
                uri=f"log:///app/{date_str}",
                name=f"App Logs - {date_str}",
                mimeType="text/plain",
                description=f"Application logs for {date_str}"
            ).model_dump()
        )

    return resources

@server.read_resource()
async def read_resource(uri: str) -> dict:
    """Read dynamic log resource."""
    if uri.startswith("log:///app/"):
        date_str = uri.removeprefix("log:///app/")
        log_content = await fetch_logs_for_date(date_str)

        return {
            "contents": [
                ResourceContent(
                    uri=uri,
                    text=log_content,
                    mimeType="text/plain"
                ).model_dump()
            ]
        }

MIME Type Detection

import mimetypes

def get_mime_type(path: Path) -> str:
    """Determine MIME type from file extension."""
    mime_type, _ = mimetypes.guess_type(str(path))

    # Fallback mapping
    if mime_type is None:
        extension_map = {
            ".md": "text/markdown",
            ".json": "application/json",
            ".yaml": "text/yaml",
            ".yml": "text/yaml",
            ".log": "text/plain",
            ".sql": "application/sql"
        }
        mime_type = extension_map.get(path.suffix, "text/plain")

    return mime_type

Security Validation

from pathlib import Path

class UriValidator:
    """Validate resource URIs for security."""

    @staticmethod
    def validate_file_uri(uri: str, allowed_roots: list[Path]) -> Path:
        """Validate file URI against allowed roots."""
        path = uri.removeprefix("file://")
        resolved = Path(path).resolve()

        # Check path traversal
        if ".." in str(path):
            raise ValueError("Path traversal not allowed")

        # Check against allowed roots
        if not any(resolved.is_relative_to(root) for root in allowed_roots):
            raise ValueError(f"Access denied: {uri}")

        return resolved

# Usage
@server.read_resource()
async def read_resource(uri: str) -> dict:
    if uri.startswith("file:///"):
        allowed_roots = [
            Path("/project/docs"),
            Path("/project/config")
        ]
        path = UriValidator.validate_file_uri(uri, allowed_roots)
        # Proceed with reading

Binary Resources

import base64

async def read_binary_resource(uri: str) -> dict:
    """Read binary resource (images, PDFs, etc.)."""
    path = Path(uri.removeprefix("file://"))

    async with aiofiles.open(path, "rb") as f:
        binary_content = await f.read()

    # Base64 encode binary data
    encoded = base64.b64encode(binary_content).decode("utf-8")

    return {
        "contents": [
            ResourceContent(
                uri=uri,
                blob=encoded,  # Use blob for binary
                mimeType=get_mime_type(path)
            ).model_dump()
        ]
    }

Database Schema Resources

@server.read_resource()
async def read_resource(uri: str) -> dict:
    """Read database schema resource."""
    if uri.startswith("db:///schemas/"):
        table_name = uri.removeprefix("db:///schemas/")

        # Fetch schema asynchronously
        schema = await fetch_table_schema(table_name)

        schema_json = {
            "table": table_name,
            "columns": [
                {
                    "name": col.name,
                    "type": col.type,
                    "nullable": col.nullable,
                    "primary_key": col.primary_key
                }
                for col in schema.columns
            ]
        }

        return {
            "contents": [
                ResourceContent(
                    uri=uri,
                    text=json.dumps(schema_json, indent=2),
                    mimeType="application/json"
                ).model_dump()
            ]
        }

Anti-Patterns

❌ Using Tools for Read-Only Operations

# WRONG - should be a resource
@server.call_tool()
async def call_tool(name, args):
    if name == "read_docs":  # This should be a resource!
        return await read_documentation()

❌ Missing URI Validation

# WRONG - security risk
path = uri.removeprefix("file://")
content = open(path).read()  # No validation!

❌ Synchronous File I/O

# WRONG - blocking operation
with open(path, "r") as f:  # Should use aiofiles
    content = f.read()

❌ Hardcoded Resource List

# WRONG - not dynamic
@server.list_resources()
async def list_resources():
    return [{"uri": "file:///README.md"}]  # Should scan filesystem

Resources (Tier 3)

MCP Resources Spec: https://spec.modelcontextprotocol.io/specification/server/resources/ aiofiles Docs: https://github.com/Tinche/aiofiles URI RFC: https://datatracker.ietf.org/doc/html/rfc3986