MCP Resources

Overview

MCP resources represent data sources, knowledge bases, and contextual information that the AI agent can access. Resources provide a standardized way to expose data through URIs with defined schemas.

Resources enable AI agents to access conversation history, documents, databases, APIs, and other data sources in a consistent manner.

Resource URIs

Resources are identified by URIs following this pattern:

{scheme}://{path}

Examples:

conversation://conv_123 - Conversation history
file:///path/to/document.txt - Local file
http://api.example.com/data - HTTP endpoint
knowledge://product-docs - Knowledge base

Resource Types

Conversation Resources

Access conversation history and context. URI Pattern: conversation://{conversation_id} Example:

## Read conversation
resource = await client.read_resource("conversation://conv_123")

content = json.loads(resource.contents[0].text)
messages = content["messages"]

for msg in messages:
    print(f"{msg['role']}: {msg['content']}")

Response:

{
  "contents": [
    {
      "uri": "conversation://conv_123",
      "mimeType": "application/json",
      "text": "{\"messages\": [{\"role\": \"user\", \"content\": \"Hello\"}, {\"role\": \"assistant\", \"content\": \"Hi!\"}], \"created_at\": \"2025-10-12T10:00:00Z\", \"updated_at\": \"2025-10-12T10:05:00Z\"}"
    }
  ]
}

File Resources

Access local or remote files. URI Pattern: file://{path} Example:

## Read file
resource = await client.read_resource("file:///docs/manual.pdf")

print(f"MIME Type: {resource.contents[0].mimeType}")
print(f"Size: {len(resource.contents[0].text)} bytes")

Response:

{
  "contents": [
    {
      "uri": "file:///docs/manual.pdf",
      "mimeType": "application/pdf",
      "blob": "base64-encoded-pdf-data"
    }
  ]
}

Knowledge Base Resources

Access curated knowledge bases and documentation. URI Pattern: knowledge://{knowledge_base_id} Example:

## Search knowledge base
resource = await client.read_resource(
    "knowledge://product-docs",
    params={"query": "installation guide"}
)

results = json.loads(resource.contents[0].text)
for doc in results["documents"]:
    print(f"Title: {doc['title']}")
    print(f"Content: {doc['content'][:200]}...\n")

Response:

{
  "contents": [
    {
      "uri": "knowledge://product-docs",
      "mimeType": "application/json",
      "text": "{\"documents\": [{\"title\": \"Installation Guide\", \"content\": \"To install...\"}, {\"title\": \"Quick Start\", \"content\": \"Getting started...\"}]}"
    }
  ]
}

HTTP Resources

Access external HTTP/HTTPS endpoints. URI Pattern: http[s]://{url} Example:

## Fetch HTTP resource
resource = await client.read_resource(
    "https://api.example.com/users/123"
)

user_data = json.loads(resource.contents[0].text)
print(f"User: {user_data['name']}")

Response:

{
  "contents": [
    {
      "uri": "https://api.example.com/users/123",
      "mimeType": "application/json",
      "text": "{\"id\": 123, \"name\": \"Alice\", \"email\": \"alice@example.com\"}"
    }
  ]
}

Database Resources

Access database tables and queries. URI Pattern: db://{database}/{table} or db://{database}/query Example:

## Read table data
resource = await client.read_resource(
    "db://analytics/events",
    params={"limit": 10, "where": "type='click'"}
)

events = json.loads(resource.contents[0].text)
print(f"Found {len(events['rows'])} events")

Response:

{
  "contents": [
    {
      "uri": "db://analytics/events",
      "mimeType": "application/json",
      "text": "{\"rows\": [{\"id\": 1, \"type\": \"click\", \"timestamp\": \"2025-10-12T10:00:00Z\"}], \"total\": 1}"
    }
  ]
}

Resource Operations

List Resources

Discover available resources:

## List all resources
resources = await client.list_resources()

for resource in resources:
    print(f"URI: {resource.uri}")
    print(f"Name: {resource.name}")
    print(f"Description: {resource.description}")
    print(f"MIME Type: {resource.mimeType}\n")

Response:

{
  "resources": [
    {
      "uri": "conversation://conv_123",
      "name": "Conversation History",
      "description": "Chat conversation with context",
      "mimeType": "application/json"
    },
    {
      "uri": "knowledge://product-docs",
      "name": "Product Documentation",
      "description": "Internal product documentation",
      "mimeType": "text/markdown"
    }
  ]
}

Read Resource

Access resource content:

## Read resource
resource = await client.read_resource("conversation://conv_123")

for content in resource.contents:
    print(f"URI: {content.uri}")
    print(f"MIME Type: {content.mimeType}")
    print(f"Content: {content.text[:200]}...")

With Parameters:

## Read with parameters
resource = await client.read_resource(
    "knowledge://product-docs",
    params={
        "query": "authentication",
        "limit": 5
    }
)

Get notifications when resources change:

## Subscribe to resource updates
await client.subscribe_resource("conversation://conv_123")

## Handle notifications
async def on_resource_updated(uri: str):
    print(f"Resource updated: {uri}")
    # Reload resource
    updated = await client.read_resource(uri)
    process_update(updated)

client.on("resource_updated", on_resource_updated)

Notification Format:

{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",
  "params": {
    "uri": "conversation://conv_123"
  }
}

Unsubscribe from Resources

Stop receiving updates:

## Unsubscribe
await client.unsubscribe_resource("conversation://conv_123")

Resource Templates

Define reusable resource patterns:

## Define template
template = {
    "uriTemplate": "conversation://{conversation_id}",
    "name": "Conversation",
    "description": "Access conversation by ID",
    "mimeType": "application/json"
}

## Use template
uri = template["uriTemplate"].format(conversation_id="conv_456")
resource = await client.read_resource(uri)

Resource Permissions

Resources respect authorization rules:

## Check access
allowed = await openfga_client.check_permission(
    user=f"user:{user_id}",
    relation="viewer",
    object="conversation:conv_123"
)

if not allowed:
    raise PermissionError("Cannot access conversation")

## Read if allowed
resource = await client.read_resource("conversation://conv_123")

Grant access:

## Grant user view access to conversation
await openfga_client.write_tuples([{
    "user": f"user:{user_id}",
    "relation": "viewer",
    "object": "conversation:conv_123"
}])

Resource Schemas

Conversation Schema

{
  "messages": [
    {
      "role": "user" | "assistant" | "system",
      "content": string,
      "timestamp": string (ISO 8601)
    }
  ],
  "conversation_id": string,
  "created_at": string (ISO 8601),
  "updated_at": string (ISO 8601),
  "metadata": {
    "user_id": string,
    "model": string,
    "total_tokens": number
  }
}

Knowledge Base Schema

{
  "documents": [
    {
      "id": string,
      "title": string,
      "content": string,
      "url": string,
      "score": number,  // Relevance score
      "metadata": object
    }
  ],
  "total_results": number,
  "query": string
}

File Schema

{
  "path": string,
  "name": string,
  "size": number,
  "mimeType": string,
  "content": string | base64,
  "created_at": string,
  "modified_at": string
}

MIME Types

Supported MIME types for resources:

MIME Type	Description	Example
`text/plain`	Plain text	Text files, logs
`text/markdown`	Markdown	Documentation
`text/html`	HTML	Web pages
`application/json`	JSON data	API responses, configs
`application/pdf`	PDF documents	Reports, manuals
`image/png`	PNG images	Screenshots, diagrams
`image/jpeg`	JPEG images	Photos
`application/octet-stream`	Binary data	Generic files

Custom Resources

Create custom resource providers:

from mcp_server_langgraph.resources import ResourceProvider

class CustomDatabaseResource(ResourceProvider):
    """Custom resource for database access"""

    def __init__(self, db_connection):
        self.db = db_connection

    async def list(self) -> list:
        """List available tables"""
        tables = await self.db.fetch("SELECT table_name FROM information_schema.tables")
        return [
            {
                "uri": f"db://mydb/{row['table_name']}",
                "name": row['table_name'],
                "description": f"Access {row['table_name']} table",
                "mimeType": "application/json"
            }
            for row in tables
        ]

    async def read(self, uri: str, params: dict = None) -> dict:
        """Read table data"""
        table_name = uri.split("/")[-1]
        limit = params.get("limit", 100) if params else 100

        rows = await self.db.fetch(
            f"SELECT * FROM {table_name} LIMIT $1",
            limit
        )

        return {
            "contents": [{
                "uri": uri,
                "mimeType": "application/json",
                "text": json.dumps({"rows": [dict(r) for r in rows]})
            }]
        }

## Register custom resource
from mcp_server_langgraph.resources import register_resource_provider

register_resource_provider("db", CustomDatabaseResource(db_connection))

Resource Caching

Cache frequently accessed resources:

from functools import lru_cache

@lru_cache(maxsize=100)
async def cached_read_resource(uri: str):
    """Read resource with caching"""
    return await client.read_resource(uri)

## Use cached version
resource = await cached_read_resource("conversation://conv_123")

Redis caching:

async def read_resource_with_cache(uri: str):
    """Read resource with Redis caching"""
    # Check cache
    cached = await redis.get(f"resource:{uri}")
    if cached:
        return json.loads(cached)

    # Fetch resource
    resource = await client.read_resource(uri)

    # Cache for 5 minutes
    await redis.setex(
        f"resource:{uri}",
        300,
        json.dumps(resource.dict())
    )

    return resource

Resource Monitoring

Track resource access:

from prometheus_client import Counter, Histogram

resource_reads = Counter(
    'mcp_resource_reads_total',
    'Total resource reads',
    ['resource_type', 'status']
)

resource_read_duration = Histogram(
    'mcp_resource_read_duration_seconds',
    'Resource read duration',
    ['resource_type']
)

## Track metrics
import time

async def tracked_read_resource(uri: str):
    resource_type = uri.split("://")[0]
    start_time = time.time()

    try:
        resource = await client.read_resource(uri)
        resource_reads.labels(
            resource_type=resource_type,
            status="success"
        ).inc()
        return resource
    except Exception as e:
        resource_reads.labels(
            resource_type=resource_type,
            status="error"
        ).inc()
        raise
    finally:
        duration = time.time() - start_time
        resource_read_duration.labels(
            resource_type=resource_type
        ).observe(duration)

Best Practices

Use Appropriate URIs

Choose descriptive, meaningful URIs:

# Good
"conversation://conv_2025-10-12_user-alice"
"knowledge://product-docs/installation"

# Bad
"resource://123"
"data://x"

Implement Pagination

For large datasets, use pagination:

page = 1
page_size = 100

while True:
    resource = await client.read_resource(
        "db://analytics/events",
        params={"page": page, "page_size": page_size}
    )

    data = json.loads(resource.contents[0].text)
    process_data(data["rows"])

    if len(data["rows"]) < page_size:
        break

    page += 1

Handle Large Files

Stream large files instead of loading fully:

# Stream file content
async for chunk in client.stream_resource("file:///large-file.bin"):
    process_chunk(chunk)

Validate Content

Validate resource content before use:

resource = await client.read_resource(uri)

# Validate MIME type
expected_type = "application/json"
if resource.contents[0].mimeType != expected_type:
    raise ValueError(f"Expected {expected_type}, got {resource.contents[0].mimeType}")

# Validate JSON structure
data = json.loads(resource.contents[0].text)
if "messages" not in data:
    raise ValueError("Invalid conversation format")

Troubleshooting

Resource not found

Error: Resource not found: conversation://conv_999Solutions:

# List available resources
resources = await client.list_resources()
available_uris = [r.uri for r in resources]
print(f"Available resources: {available_uris}")

# Check if resource exists before reading
if uri in available_uris:
    resource = await client.read_resource(uri)

Permission denied

Error: User does not have permission to read resourceSolutions:

# Check permission
resource_id = uri.split("://")[1]
allowed = await openfga_client.check_permission(
    user=f"user:{user_id}",
    relation="viewer",
    object=f"conversation:{resource_id}"
)

if not allowed:
    # Request access or show error
    print("Access denied. Contact owner for permission.")

Invalid URI format

Error: Invalid resource URI formatSolutions:

# Validate URI format
import re

uri_pattern = r"^[a-z]+://[a-zA-Z0-9_\-/]+$"

if not re.match(uri_pattern, uri):
    raise ValueError(f"Invalid URI format: {uri}")

# Parse URI components
scheme, path = uri.split("://", 1)
print(f"Scheme: {scheme}, Path: {path}")

Next Steps

MCP Tools

Available tools

MCP Messages

Message protocol

MCP Endpoints

HTTP endpoints

Authorization

Resource permissions

MCP Resources Ready: Standardized access to data sources and context!

API Documentation

Developer Tools

MCP Protocol

Overview

Resource URIs

Resource Types

Conversation Resources

File Resources

Knowledge Base Resources

HTTP Resources

Database Resources

Resource Operations

List Resources

Read Resource

Unsubscribe from Resources

Resource Templates

Resource Permissions

Resource Schemas

Conversation Schema

Knowledge Base Schema

File Schema

MIME Types

Custom Resources

Resource Caching

Resource Monitoring

Best Practices

Troubleshooting

Next Steps

MCP Tools

MCP Messages

MCP Endpoints

Authorization

API Documentation

Developer Tools

MCP Protocol

​Overview

​Resource URIs

​Resource Types

​Conversation Resources

​File Resources

​Knowledge Base Resources

​HTTP Resources

​Database Resources

​Resource Operations

​List Resources

​Read Resource

​Subscribe to Resources

​Unsubscribe from Resources

​Resource Templates

​Resource Permissions

​Resource Schemas

​Conversation Schema

​Knowledge Base Schema

​File Schema

​MIME Types

​Custom Resources

​Resource Caching

​Resource Monitoring

​Best Practices

​Troubleshooting

​Next Steps

MCP Tools

MCP Messages

MCP Endpoints

Authorization

Overview

Resource URIs

Resource Types

Conversation Resources

File Resources

Knowledge Base Resources

HTTP Resources

Database Resources

Resource Operations

List Resources

Read Resource

Subscribe to Resources

Unsubscribe from Resources

Resource Templates

Resource Permissions

Resource Schemas

Conversation Schema

Knowledge Base Schema

File Schema

MIME Types

Custom Resources

Resource Caching

Resource Monitoring

Best Practices

Troubleshooting

Next Steps