Implement MCP resources

Introduction

When building MCP servers, exposing data to an LLM requires more than just returning strings — clients need to discover what resources exist, fetch them by a stable URI, and receive updates when content changes. Without a consistent resource abstraction, each server invents its own ad-hoc approach, making integrations brittle and discovery impossible. In this lesson you will implement a complete ResourceProvider that registers ResourceDefinition objects, serves content through async handlers, manages subscriber callbacks for live updates, and constructs well-formed URIs for both file-based and custom-scheme resources.

Key Terminology

• ResourceDefinition — A Python dataclass that bundles all metadata for a single MCP resource: its uri, human-readable name, description, mime_type, and an async handler callable that fetches content on demand.
• ResourceProvider — The central registry class that stores ResourceDefinition objects keyed by URI, dispatches read() calls to the appropriate handler, and manages per-URI subscriber callback lists via _resources and _subscriptions dictionaries.
• Resource URI — A stable, unique string identifier for an MCP resource, constructed either with build_file_uri() for filesystem paths or build_custom_uri() for application-defined schemes; urllib.parse percent-encoding ensures the URI is safe for transport.
• Async handler — An Optional[Callable[[], Awaitable[str]]] field on ResourceDefinition that ResourceProvider.read() awaits to retrieve current resource content without blocking the event loop.
• Observer pattern — The subscription mechanism implemented by subscribe(), unsubscribe(), and notify_update() that lets clients register async callbacks to receive pushed content whenever a resource changes, without the server needing to know which clients are listening.
• MIME type — A metadata field on ResourceDefinition (defaulting to "text/plain") that is forwarded in the MCP Resource protocol object so clients know how to interpret the content they receive from a read() call.

Concepts

Resources as a Discoverable Contract

MCP resources give an LLM structured, URI-addressed access to data — configuration files, live metrics, database snapshots, or any content that enriches a prompt. The key design principle is that discovery and fetching are separated: get_resources() returns a list of lightweight Resource protocol objects (URI, name, description, MIME type) that a client can inspect without triggering any I/O. Only when a client explicitly calls read() does the provider invoke the async handler and fetch real content.

This separation lets a server advertise many resources at negotiation time while clients selectively pull only what their current task needs. The to_mcp_resource() method on ResourceDefinition handles the translation from the internal dataclass to the wire-level Resource type expected by the MCP SDK, keeping protocol concerns out of the domain model (see Code Walkthrough).

On-Demand Fetching and Explicit Error Boundaries

Fetching is deferred to an async handler callable rather than storing content directly on the ResourceDefinition. This matters because resources often represent live data — a file that changes, an API response, a queue depth — not a value known at registration time. Storing a callable means ResourceProvider invokes it at read time and always delivers fresh content.

read() enforces two distinct error boundaries: it raises ValueError for an unrecognized URI (the resource was never registered) and a separate ValueError when a registered resource has no handler (a definition-only placeholder). Both are explicit failures — the caller always knows why a read did not succeed rather than receiving None or a silent empty string.

Push Notifications via the Observer Pattern

Static reads cover most use cases, but some clients need to react to resource changes without polling. ResourceProvider implements the observer pattern through three methods: subscribe() appends a per-URI async callback to _subscriptions; unsubscribe() removes it by identity comparison; and notify_update() calls read() once to get current content, then fans it out to every registered callback in order.

This design keeps notification logic centralized — the server calls notify_update(uri) whenever content changes, and all interested clients receive the new value without knowing about each other. Crucially, the same async handler that serves explicit reads also serves the push path, so there is no duplication of fetch logic (see Code Walkthrough).

URI Construction Conventions

Every resource needs a URI that is stable, unique, and safe to embed in protocol messages. build_file_uri() wraps urllib.parse.quote() around a filesystem path to produce a valid file:// URI, handling spaces and special characters through percent-encoding. build_custom_uri() extends this to arbitrary application-defined schemes — db://, metrics://, or any custom prefix — and optionally appends query parameters via urllib.parse.urlencode(). Using these helpers instead of manual string concatenation ensures that edge-case characters in paths or parameter values never produce a malformed URI that clients cannot parse or dereference.

Code Walkthrough

Now that you've seen Resources as a Discoverable Contract, On-Demand Fetching and Explicit Error Boundaries, Push Notifications via the Observer Pattern, and URI Construction Conventions, this walkthrough turns them into working code.

Resource Structure and URIs

Resources provide read-only access to data that can enrich LLM context. Each resource is identified by a URI and has associated metadata including name, description, and MIME type. The following code defines a ResourceDefinition dataclass that captures all resource metadata alongside an async handler for fetching content. It also introduces the ResourceProvider class that manages resource registrations, handles read requests by delegating to the appropriate handler, and supports a subscription mechanism for notifying clients when resource content changes.

1from dataclasses import dataclass, field
2from typing import Dict, Any, List, Optional, Callable, Awaitable
3from mcp.types import Resource, TextResourceContents, BlobResourceContents
4import urllib.parse
5
6@dataclass
7class ResourceDefinition:
8    """
9    Definition of an MCP resource.
10
11    Attributes:
12        uri: Unique resource identifier (URI format)
13        name: Human-readable resource name
14        description: Description of the resource content
15        mime_type: MIME type of the content
16        handler: Async function that fetches the resource
17    """
18    uri: str
19    name: str
20    description: str
21    mime_type: str = "text/plain"
22    handler: Optional[Callable[[], Awaitable[str]]] = None
23
24    def to_mcp_resource(self) -> Resource:
25        """Convert to MCP Resource type."""
26        return Resource(
27            uri=self.uri,
28            name=self.name,
29            description=self.description,
30            mimeType=self.mime_type
31        )
32
33class ResourceProvider:
34    """
35    Provides access to MCP resources.
36
37    Manages resource definitions, handles read requests,
38    and supports resource subscriptions.
39    """

The continuation below implements the ResourceProvider methods in full. The constructor initializes dictionaries for resource storage and subscription tracking. Registration, listing, and reading methods provide the core interface, while subscribe, unsubscribe, and notify_update implement the observer pattern for resource change notifications. Utility functions at the end construct properly formatted file URIs and custom-scheme URIs with optional query parameters.

1    def __init__(self):
2        self._resources: Dict[str, ResourceDefinition] = {}
3        self._subscriptions: Dict[str, List[Callable]] = {}
4
5    def register(self, resource: ResourceDefinition) -> None:
6        """Register a resource definition."""
7        self._resources[resource.uri] = resource
8
9    def get_resources(self) -> List[Resource]:
10        """Get all registered resources in MCP format."""
11        return [r.to_mcp_resource() for r in self._resources.values()]
12
13    async def read(self, uri: str) -> str:
14        resource = self._resources.get(uri)
15        if not resource:
16            raise ValueError(f"Unknown resource: {uri}")
17        if resource.handler:
18            return await resource.handler()
19        raise ValueError(f"Resource has no handler: {uri}")
20
21    def subscribe(self, uri: str, callback: Callable[[str], Awaitable[None]]) -> None:
22        if uri not in self._subscriptions:
23            self._subscriptions[uri] = []
24        self._subscriptions[uri].append(callback)
25
26    def unsubscribe(self, uri: str, callback: Callable[[str], Awaitable[None]]) -> None:
27        if uri in self._subscriptions:
28            self._subscriptions[uri] = [cb for cb in self._subscriptions[uri] if cb != callback]
29
30    async def notify_update(self, uri: str) -> None:
31        if uri in self._subscriptions:
32            content = await self.read(uri)
33            for callback in self._subscriptions[uri]:
34                await callback(content)
35
36def build_file_uri(path: str) -> str:
37    return f"file://{urllib.parse.quote(path)}"
38
39def build_custom_uri(scheme: str, path: str, params: Dict[str, str] = None) -> str:
40    uri = f"{scheme}://{path}"
41    if params:
42        uri = f"{uri}?{urllib.parse.urlencode(params)}"
43    return uri

To verify the implementation is wired correctly, instantiate ResourceProvider, call register() with a ResourceDefinition whose handler returns a known string, then assert that get_resources() returns a list containing that URI and that await provider.read(uri) returns the expected string without raising.

Do's and Don'ts

Having walked through implementing resources above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do provide an async handler on every ResourceDefinition you intend to serve — ResourceProvider.read() awaits resource.handler() to fetch content; a definition registered without a handler raises ValueError("Resource has no handler") at read time even though get_resources() still advertises the URI as discoverable.
2. ✓Do use build_file_uri() and build_custom_uri() instead of raw f-string concatenation when constructing resource URIs — both helpers call urllib.parse.quote() or urllib.parse.urlencode(), which percent-encodes spaces and special characters that would otherwise produce malformed URIs clients cannot dereference.
3. ✓Do call subscribe() with a stable, named callback object rather than an inline lambda when you will later need to unsubscribe() — unsubscribe() removes callbacks by object identity (cb != callback); a fresh lambda created at each call site is a distinct object and will never match, leaving the callback permanently registered and receiving notifications after the caller has moved on.

Don'ts

1. ✗Don't return ResourceDefinition objects directly from get_resources() — the MCP client expects Resource protocol objects (from mcp.types), whose field is mimeType not mime_type; the to_mcp_resource() conversion method exists specifically to produce the correct wire shape, and skipping it causes schema validation failures on the client side.
2. ✗Don't call notify_update() before subscribers have been registered via subscribe() — notify_update() fans out only to callbacks already present in _subscriptions[uri] at call time; any client that subscribes after the update fires receives no notification and sees stale content until the next change.
3. ✗Don't suppress or swallow the ValueError that read() raises for an unknown URI — that exception is the only signal that a URI was never passed to register(); silently returning an empty string or None in a caller-side except block hides registration bugs and causes the LLM context to be silently enriched with empty or wrong data.