mcp-resource-exposure-patterns

What this skill does

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

```python
# 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

```python
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

```python
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

```python
@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

```python
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

```python
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

```python
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

```python
@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
            ]
        }