ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 07:09:11 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add PromptsAsTools transform (#2946)

Browse source
This commit is contained in:
Jeremiah Lowin 2026-01-19 18:19:53 -05:00 • committed by GitHub
commit c0ef90e713
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 2767 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

50
docs/development/v3-notes/v3-features.mdx
View file

@ -108,6 +108,8 @@ Transforms modify components (tools, resources, prompts) as they flow from provi
- `ToolTransform` - modifies tool schemas (rename, description, tags, argument transforms)
- `Enabled` - sets enabled state on components by key or tag (backs `enable()`/`disable()` API)
- `VersionFilter` - filters components by version range (`version_gte`, `version_lt`)
- `ResourcesAsTools` - exposes resources as tools for tool-only clients
- `PromptsAsTools` - exposes prompts as tools for tool-only clients
```python
from fastmcp.server.transforms import Namespace, ToolTransform
@ -149,6 +151,54 @@ Transforms apply at two levels:
Documentation: `docs/servers/providers/transforms.mdx`, `docs/servers/enabled.mdx`
### ResourcesAsTools and PromptsAsTools
These transforms expose resources and prompts as tools for clients that only support the tools protocol. Each transform generates two tools that provide listing and access functionality.
**ResourcesAsTools** generates `list_resources` and `read_resource` tools:
```python
from fastmcp import FastMCP
from fastmcp.server.transforms import ResourcesAsTools
mcp = FastMCP("Server")
@mcp.resource("data://config")
def get_config() -> dict:
return {"setting": "value"}
mcp.add_transform(ResourcesAsTools(mcp))
# Now has list_resources and read_resource tools
```
The `list_resources` tool returns JSON with resource metadata. The `read_resource` tool accepts a URI and returns the resource content, preserving both text and binary data through base64 encoding.
**PromptsAsTools** generates `list_prompts` and `get_prompt` tools:
```python
from fastmcp import FastMCP
from fastmcp.server.transforms import PromptsAsTools
mcp = FastMCP("Server")
@mcp.prompt
def analyze_code(code: str, language: str = "python") -> str:
return f"Analyze this {language} code:\n{code}"
mcp.add_transform(PromptsAsTools(mcp))
# Now has list_prompts and get_prompt tools
```
The `list_prompts` tool returns JSON with prompt metadata including argument information. The `get_prompt` tool accepts a prompt name and optional arguments dict, returning the rendered prompt as a messages array. Non-text content (like embedded resources) is preserved as structured JSON.
Both transforms:
- Capture a provider reference at construction for deferred querying
- Route through `FastMCP.read_resource()` / `FastMCP.render_prompt()` when the provider is FastMCP, ensuring middleware chains execute
- Fall back to direct provider methods for plain providers
- Return JSON for easy parsing by tool-only clients
Documentation: `docs/servers/providers/resources-as-tools.mdx`, `docs/servers/providers/prompts-as-tools.mdx`
---
## Session-Scoped State

1
docs/docs.json
View file

@ -110,6 +110,7 @@
"servers/providers/overview",
"servers/providers/transforms",
"servers/providers/resources-as-tools",
"servers/providers/prompts-as-tools",
"servers/providers/local",
"servers/providers/filesystem",
"servers/providers/mounting",

125
docs/servers/providers/prompts-as-tools.mdx Normal file
View file

@ -0,0 +1,125 @@
---
title: Prompts as Tools
sidebarTitle: Prompts as Tools
description: Expose prompts to tool-only clients
icon: message-lines
---
import { VersionBadge } from '/snippets/version-badge.mdx'
<VersionBadge version="3.0.0" />
Some MCP clients only support tools. They cannot list or get prompts directly because they lack prompt protocol support. The `PromptsAsTools` transform bridges this gap by generating tools that provide access to your server's prompts.
When you add `PromptsAsTools` to a server, it creates two tools that clients can call instead of using the prompt protocol:
- **`list_prompts`** returns JSON describing all available prompts and their arguments
- **`get_prompt`** renders a specific prompt with provided arguments
This means any client that can call tools can now access prompts, even if the client has no native prompt support.
## Basic Usage
Pass your server to `PromptsAsTools` when adding the transform. The transform queries that server for prompts whenever the generated tools are called.
```python
from fastmcp import FastMCP
from fastmcp.server.transforms import PromptsAsTools
mcp = FastMCP("My Server")
@mcp.prompt
def analyze_code(code: str, language: str = "python") -> str:
"""Analyze code for potential issues."""
return f"Analyze this {language} code:\n{code}"
@mcp.prompt
def explain_concept(concept: str) -> str:
"""Explain a programming concept."""
return f"Explain: {concept}"
# Add the transform - creates list_prompts and get_prompt tools
mcp.add_transform(PromptsAsTools(mcp))
```
Clients now see three items: whatever tools you defined directly, plus `list_prompts` and `get_prompt`.
## Listing Prompts
The `list_prompts` tool returns JSON with metadata for each prompt, including its arguments.
```python
result = await client.call_tool("list_prompts", {})
prompts = json.loads(result.data)
# [
# {
# "name": "analyze_code",
# "description": "Analyze code for potential issues.",
# "arguments": [
# {"name": "code", "description": null, "required": true},
# {"name": "language", "description": null, "required": false}
# ]
# },
# {
# "name": "explain_concept",
# "description": "Explain a programming concept.",
# "arguments": [
# {"name": "concept", "description": null, "required": true}
# ]
# }
#]
```
Each argument includes:
- `name`: The argument name
- `description`: Optional description from type hints or docstrings
- `required`: Whether the argument must be provided
## Getting Prompts
The `get_prompt` tool accepts a prompt name and optional arguments dict. It returns the rendered prompt as JSON with a messages array.
```python
# Prompt with required and optional arguments
result = await client.call_tool(
"get_prompt",
{
"name": "analyze_code",
"arguments": {
"code": "x = 1\nprint(x)",
"language": "python"
}
}
)
response = json.loads(result.data)
# {
# "messages": [
# {
# "role": "user",
# "content": "Analyze this python code:\nx = 1\nprint(x)"
# }
# ]
# }
```
If a prompt has no arguments, you can omit the `arguments` field or pass an empty dict:
```python
result = await client.call_tool(
"get_prompt",
{"name": "simple_prompt"}
)
```
## Message Format
Rendered prompts return a messages array following the standard MCP format. Each message includes:
- `role`: The message role (typically "user", "assistant", or "system")
- `content`: The message text content
Multi-message prompts are supported - the array will contain all messages in order.
## Binary Content
Unlike resources, prompts always return text content. There is no binary encoding needed.

77
examples/prompts_as_tools/client.py Normal file
View file

@ -0,0 +1,77 @@
"""Example: Client using prompts-as-tools.
This client demonstrates calling the list_prompts and get_prompt tools
generated by the PromptsAsTools transform.
Run with:
uv run python examples/prompts_as_tools/client.py
"""
import asyncio
import json
from fastmcp.client import Client
async def main():
# Connect to the server
async with Client("examples/prompts_as_tools/server.py") as client:
# List all available tools
print("=== Available Tools ===")
tools = await client.list_tools()
for tool in tools:
print(f" - {tool.name}: {tool.description}")
print()
# Use list_prompts tool to see what's available
print("=== Listing Prompts ===")
result = await client.call_tool("list_prompts", {})
prompts = json.loads(result.data)
for prompt in prompts:
print(f" {prompt['name']}")
print(f" Description: {prompt.get('description', 'N/A')}")
if prompt["arguments"]:
print(" Arguments:")
for arg in prompt["arguments"]:
required = "required" if arg["required"] else "optional"
print(
f" - {arg['name']} ({required}): {arg.get('description', 'N/A')}"
)
print()
# Get a prompt without optional arguments
print("=== Getting Simple Prompt ===")
result = await client.call_tool(
"get_prompt",
{"name": "explain_concept", "arguments": {"concept": "recursion"}},
)
response = json.loads(result.data)
print("Messages:")
for msg in response["messages"]:
print(f" Role: {msg['role']}")
print(f" Content: {msg['content'][:100]}...")
print()
# Get a prompt with optional arguments
print("=== Getting Prompt with Optional Arguments ===")
result = await client.call_tool(
"get_prompt",
{
"name": "analyze_code",
"arguments": {
"code": "def factorial(n):\n return n * factorial(n-1)",
"language": "python",
"focus": "bugs",
},
},
)
response = json.loads(result.data)
print("Messages:")
for msg in response["messages"]:
print(f" Role: {msg['role']}")
print(f" Content: {msg['content'][:150]}...")
if __name__ == "__main__":
asyncio.run(main())

86
examples/prompts_as_tools/server.py Normal file
View file

@ -0,0 +1,86 @@
"""Example: Expose prompts as tools using PromptsAsTools transform.
This example shows how to use PromptsAsTools to make prompts accessible
to clients that only support tools (not the prompts protocol).
Run with:
uv run python examples/prompts_as_tools/server.py
"""
from fastmcp import FastMCP
from fastmcp.server.transforms import PromptsAsTools
mcp = FastMCP("Prompt Tools Demo")
# Simple prompt without arguments
@mcp.prompt
def explain_concept(concept: str) -> str:
"""Explain a programming concept."""
return f"""Please explain the following programming concept in simple terms:
{concept}
Include:
- A clear definition
- Common use cases
- A simple example
"""
# Prompt with multiple arguments
@mcp.prompt
def analyze_code(code: str, language: str = "python", focus: str = "all") -> str:
"""Analyze code for potential issues."""
return f"""Analyze this {language} code:
```{language}
{code}
```
Focus on: {focus}
Please identify:
- Potential bugs or errors
- Performance issues
- Code style improvements
- Security concerns
"""
# Prompt with required and optional arguments
@mcp.prompt
def review_pull_request(
title: str, description: str, diff: str, guidelines: str = ""
) -> str:
"""Review a pull request."""
guidelines_section = (
f"\n\nGuidelines to follow:\n{guidelines}" if guidelines else ""
)
return f"""Review this pull request:
**Title:** {title}
**Description:**
{description}
**Diff:**
```
{diff}
```{guidelines_section}
Please provide:
- Summary of changes
- Potential issues or concerns
- Suggestions for improvement
- Overall recommendation (approve/request changes)
"""
# Add the transform - this creates list_prompts and get_prompt tools
mcp.add_transform(PromptsAsTools(mcp))
if __name__ == "__main__":
mcp.run()

387
plans/03-health-endpoint.md Normal file
View file

@ -0,0 +1,387 @@
# Health Endpoint
## Problem
Production deployments need health checks for:
- Load balancers (AWS ALB, nginx, etc.)
- Kubernetes liveness/readiness probes
- Monitoring systems (Datadog, Prometheus, etc.)
- Uptime monitoring services
Currently users must implement health endpoints themselves or rely on generic HTTP probes that don't understand server state.
## Solution
Built-in `/health` endpoint that returns server status and optional custom health checks.
## API
### Basic Usage
```python
from fastmcp import FastMCP
# Enable health endpoint (defaults to /health)
mcp = FastMCP("server", health_endpoint=True)
```
### With Configuration
```python
from fastmcp.server.health import HealthConfig, HealthCheck
async def check_database() -> tuple[str, bool]:
"""Custom health check."""
try:
await db.execute("SELECT 1")
return "database", True
except Exception:
return "database", False
async def check_cache() -> tuple[str, bool]:
try:
await cache.ping()
return "cache", True
except Exception:
return "cache", False
mcp = FastMCP(
"server",
health_endpoint=HealthConfig(
path="/health",
checks=[check_database, check_cache],
include_version=True,
include_uptime=True,
)
)
```
### Readiness Endpoint
```python
# Separate liveness (is server running?) and readiness (can it serve traffic?)
mcp = FastMCP(
"server",
health_endpoint=HealthConfig(
liveness_path="/health",
readiness_path="/ready",
readiness_checks=[check_database, check_cache],
)
)
```
## Response Format
### Healthy Response (200 OK)
```json
{
"status": "healthy",
"version": "3.0.0",
"uptime_seconds": 3600.5,
"timestamp": "2025-01-19T12:00:00Z",
"checks": {
"database": "ok",
"cache": "ok"
}
}
```
### Unhealthy Response (503 Service Unavailable)
```json
{
"status": "unhealthy",
"version": "3.0.0",
"uptime_seconds": 3600.5,
"timestamp": "2025-01-19T12:00:00Z",
"checks": {
"database": "failed",
"cache": "ok"
}
}
```
### Minimal Response
If `checks=[]`, `include_version=False`, `include_uptime=False`:
```json
{
"status": "healthy"
}
```
## Implementation
### Location
- `src/fastmcp/server/health.py` - Core implementation
- `src/fastmcp/server/server.py` - Integration with FastMCP
### HealthConfig
```python
from dataclasses import dataclass
from collections.abc import Callable, Awaitable
HealthCheckFn = Callable[[], Awaitable[tuple[str, bool]]]
@dataclass
class HealthConfig:
"""Configuration for health endpoints."""
# Liveness endpoint (is server alive?)
liveness_path: str = "/health"
# Readiness endpoint (can server handle requests?)
readiness_path: str | None = None
# Health checks (name, passed/failed)
checks: list[HealthCheckFn] = field(default_factory=list)
# Checks only for readiness (not liveness)
readiness_checks: list[HealthCheckFn] = field(default_factory=list)
# Include server version in response
include_version: bool = True
# Include uptime in response
include_uptime: bool = True
# Timeout for each check
check_timeout: float = 5.0
```
### Health Endpoint Handler
```python
class HealthHandler:
def __init__(self, config: HealthConfig, server: FastMCP):
self.config = config
self.server = server
self.start_time = time.time()
async def liveness(self, request: Request) -> Response:
"""Liveness check - is the server running?"""
result = {
"status": "healthy",
}
if self.config.include_version:
result["version"] = fastmcp.__version__
if self.config.include_uptime:
result["uptime_seconds"] = time.time() - self.start_time
if self.config.checks:
check_results = await self._run_checks(self.config.checks)
result["checks"] = check_results
if any(status == "failed" for status in check_results.values()):
result["status"] = "unhealthy"
return JSONResponse(result, status_code=503)
result["timestamp"] = datetime.now(timezone.utc).isoformat()
return JSONResponse(result)
async def readiness(self, request: Request) -> Response:
"""Readiness check - can the server handle requests?"""
# Liveness checks
liveness_checks = await self._run_checks(self.config.checks)
# Readiness-specific checks
readiness_checks = await self._run_checks(self.config.readiness_checks)
all_checks = {**liveness_checks, **readiness_checks}
result = {
"status": "ready" if all(s == "ok" for s in all_checks.values()) else "not_ready",
"checks": all_checks,
}
if self.config.include_version:
result["version"] = fastmcp.__version__
if self.config.include_uptime:
result["uptime_seconds"] = time.time() - self.start_time
result["timestamp"] = datetime.now(timezone.utc).isoformat()
status_code = 200 if result["status"] == "ready" else 503
return JSONResponse(result, status_code=status_code)
async def _run_checks(self, checks: list[HealthCheckFn]) -> dict[str, str]:
"""Run health checks with timeout."""
results = {}
for check in checks:
try:
async with asyncio.timeout(self.config.check_timeout):
name, passed = await check()
results[name] = "ok" if passed else "failed"
except asyncio.TimeoutError:
results[name] = "timeout"
except Exception:
results[name] = "error"
return results
```
### Integration with FastMCP
In `server.py`:
```python
class FastMCP:
def __init__(
self,
name: str,
*,
health_endpoint: bool | HealthConfig = False,
**kwargs,
):
self._health_handler: HealthHandler | None = None
if health_endpoint:
config = health_endpoint if isinstance(health_endpoint, HealthConfig) else HealthConfig()
self._health_handler = HealthHandler(config, self)
async def handle_http_request(self, request: Request) -> Response:
"""Route HTTP requests."""
path = request.url.path
# Health endpoints
if self._health_handler:
if path == self._health_handler.config.liveness_path:
return await self._health_handler.liveness(request)
if self._health_handler.config.readiness_path and path == self._health_handler.config.readiness_path:
return await self._health_handler.readiness(request)
# Normal MCP request handling
return await self._handle_mcp_request(request)
```
## Edge Cases
1. **stdio transport** - Health endpoints only work for HTTP transports. Ignore for stdio.
2. **Startup time** - Readiness checks might fail during startup (DB connecting, etc.). This is correct behavior.
3. **Check timeouts** - Individual checks should timeout independently. Don't let one slow check block others.
4. **Check errors** - Exceptions in checks should be caught and reported as "error" status.
5. **No checks** - If no checks configured, health endpoint just returns `{"status": "healthy"}`.
6. **Concurrent requests** - Health checks might be called concurrently. Ensure checks are safe for concurrent execution.
## Testing
Add `tests/server/test_health.py`:
```python
async def test_basic_health_endpoint():
mcp = FastMCP("test", health_endpoint=True)
async with httpx.AsyncClient(app=mcp.get_asgi_app()) as client:
response = await client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
async def test_health_checks():
async def good_check():
return "good", True
async def bad_check():
return "bad", False
mcp = FastMCP(
"test",
health_endpoint=HealthConfig(checks=[good_check, bad_check])
)
async with httpx.AsyncClient(app=mcp.get_asgi_app()) as client:
response = await client.get("/health")
assert response.status_code == 503
data = response.json()
assert data["status"] == "unhealthy"
assert data["checks"]["good"] == "ok"
assert data["checks"]["bad"] == "failed"
async def test_readiness_endpoint():
async def db_check():
return "database", True
mcp = FastMCP(
"test",
health_endpoint=HealthConfig(
readiness_path="/ready",
readiness_checks=[db_check]
)
)
async with httpx.AsyncClient(app=mcp.get_asgi_app()) as client:
# Liveness should pass with no checks
response = await client.get("/health")
assert response.status_code == 200
# Readiness should pass with passing checks
response = await client.get("/ready")
assert response.status_code == 200
assert response.json()["status"] == "ready"
```
## Documentation
Add to `docs/servers/health.mdx`:
- Why health endpoints matter
- Basic usage
- Custom health checks
- Liveness vs readiness
- Kubernetes integration examples
- AWS ALB integration examples
- Best practices for health checks
## Examples
### Kubernetes Integration
```yaml
apiVersion: v1
kind: Pod
metadata:
name: fastmcp-server
spec:
containers:
- name: server
image: my-fastmcp-server
ports:
- containerPort: 8000
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 30
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
```
### AWS ALB Target Group
Health check configuration:
- Protocol: HTTP
- Path: `/health`
- Success codes: 200
- Interval: 30 seconds
- Timeout: 5 seconds
- Healthy threshold: 2
- Unhealthy threshold: 3

484
plans/04-resource-subscriptions.md Normal file
View file

@ -0,0 +1,484 @@
# Resource Subscriptions
## Problem
The MCP spec supports resource subscriptions where:
1. Client subscribes to a resource URI
2. Server notifies client when that resource changes
3. Client can then re-read the resource
This enables real-time updates without polling. FastMCP doesn't currently implement this.
## Solution
Full implementation of MCP resource subscription protocol.
## API
### Server-side: Declaring Subscribable Resources
```python
from fastmcp import FastMCP, Context
mcp = FastMCP("server")
# Option 1: Decorator parameter
@mcp.resource("data://metrics", subscribe=True)
async def get_metrics() -> dict:
return {"cpu": 45.2, "memory": 60.1}
# Option 2: ResourceConfig
from fastmcp.resources import ResourceConfig
@mcp.resource(
"data://metrics",
config=ResourceConfig(subscribable=True)
)
async def get_metrics() -> dict:
return {"cpu": 45.2, "memory": 60.1}
```
### Server-side: Notifying Subscribers
```python
@mcp.tool
async def update_metrics(ctx: Context, cpu: float, memory: float) -> str:
global current_metrics
current_metrics = {"cpu": cpu, "memory": memory}
# Notify all subscribers that this resource changed
await ctx.notify_resource_changed("data://metrics")
return "Metrics updated"
# Or notify from anywhere with access to server
async def background_updater(server: FastMCP):
while True:
await asyncio.sleep(5)
# Update data...
await server.notify_resource_changed("data://metrics")
```
### Client-side: Subscribing
```python
from fastmcp import Client
async with Client("http://server:8000/mcp") as client:
# Subscribe to a resource
await client.subscribe_resource("data://metrics")
# Handle notifications
async for notification in client.resource_notifications():
print(f"Resource changed: {notification.uri}")
# Re-read the resource
updated_data = await client.read_resource(notification.uri)
print(f"New data: {updated_data}")
```
### Client-side: Unsubscribing
```python
async with Client("http://server:8000/mcp") as client:
# Subscribe
await client.subscribe_resource("data://metrics")
# Later, unsubscribe
await client.unsubscribe_resource("data://metrics")
```
### Client-side: List Subscriptions
```python
async with Client("http://server:8000/mcp") as client:
await client.subscribe_resource("data://metrics")
await client.subscribe_resource("data://logs")
# Get all active subscriptions
subs = client.list_subscriptions()
assert "data://metrics" in subs
assert "data://logs" in subs
```
## MCP Protocol
### Subscribe Request
```json
{
"method": "resources/subscribe",
"params": {
"uri": "data://metrics"
}
}
```
### Subscribe Response
```json
{
"result": {}
}
```
### Unsubscribe Request
```json
{
"method": "resources/unsubscribe",
"params": {
"uri": "data://metrics"
}
}
```
### Resource Changed Notification
Server → Client notification:
```json
{
"method": "notifications/resources/updated",
"params": {
"uri": "data://metrics"
}
}
```
## Implementation
### Location
- `src/fastmcp/server/subscriptions.py` - Server-side subscription tracking
- `src/fastmcp/server/context.py` - Add `notify_resource_changed()` method
- `src/fastmcp/server/server.py` - Wire up subscription handlers
- `src/fastmcp/client/client.py` - Client subscription API
- `src/fastmcp/resources/resource.py` - Add `subscribe` parameter
### Server-side: Subscription Manager
```python
from collections import defaultdict
from weakref import WeakSet
class SubscriptionManager:
"""Tracks resource subscriptions per session."""
def __init__(self):
# uri -> set of session IDs
self._subscriptions: dict[str, set[str]] = defaultdict(set)
# session_id -> set of URIs (for cleanup)
self._session_subscriptions: dict[str, set[str]] = defaultdict(set)
def subscribe(self, uri: str, session_id: str) -> None:
"""Subscribe a session to a resource URI."""
self._subscriptions[uri].add(session_id)
self._session_subscriptions[session_id].add(uri)
def unsubscribe(self, uri: str, session_id: str) -> None:
"""Unsubscribe a session from a resource URI."""
self._subscriptions[uri].discard(session_id)
self._session_subscriptions[session_id].discard(uri)
def get_subscribers(self, uri: str) -> set[str]:
"""Get all session IDs subscribed to a URI."""
return self._subscriptions.get(uri, set()).copy()
def cleanup_session(self, session_id: str) -> None:
"""Remove all subscriptions for a session."""
for uri in self._session_subscriptions.get(session_id, set()):
self._subscriptions[uri].discard(session_id)
del self._session_subscriptions[session_id]
def list_subscriptions(self, session_id: str) -> list[str]:
"""List all URIs a session is subscribed to."""
return list(self._session_subscriptions.get(session_id, set()))
```
### Server-side: Context Method
```python
class Context:
async def notify_resource_changed(self, uri: str) -> None:
"""Notify all subscribers that a resource has changed."""
manager = self._server._subscription_manager
session_ids = manager.get_subscribers(uri)
for session_id in session_ids:
session = self._server._get_session(session_id)
if session:
await session.send_notification(
"notifications/resources/updated",
{"uri": uri}
)
```
### Server-side: Session Notification
```python
class Session:
async def send_notification(self, method: str, params: dict) -> None:
"""Send a notification to the client."""
message = {
"jsonrpc": "2.0",
"method": method,
"params": params,
}
if self.transport == "stdio":
await self._stdio_write(message)
elif self.transport in ("sse", "streamable-http"):
await self._http_write_notification(message)
```
### Server-side: Handlers
In `server.py`:
```python
async def handle_subscribe(self, uri: str, ctx: Context) -> dict:
"""Handle resources/subscribe request."""
# Check if resource exists and is subscribable
resource = await self.get_resource(uri)
if not resource:
raise McpError(f"Resource not found: {uri}", code=-32602)
if not resource.subscribable:
raise McpError(f"Resource not subscribable: {uri}", code=-32602)
# Add subscription
self._subscription_manager.subscribe(uri, ctx.session_id)
return {}
async def handle_unsubscribe(self, uri: str, ctx: Context) -> dict:
"""Handle resources/unsubscribe request."""
self._subscription_manager.unsubscribe(uri, ctx.session_id)
return {}
```
### Client-side: Subscription API
```python
class Client:
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._subscriptions: set[str] = set()
self._notification_queue: asyncio.Queue = asyncio.Queue()
async def subscribe_resource(self, uri: str) -> None:
"""Subscribe to resource change notifications."""
result = await self.request(
"resources/subscribe",
{"uri": uri}
)
self._subscriptions.add(uri)
async def unsubscribe_resource(self, uri: str) -> None:
"""Unsubscribe from resource change notifications."""
await self.request(
"resources/unsubscribe",
{"uri": uri}
)
self._subscriptions.discard(uri)
def list_subscriptions(self) -> list[str]:
"""List all active subscriptions."""
return list(self._subscriptions)
async def resource_notifications(self) -> AsyncIterator[ResourceNotification]:
"""Iterate over resource change notifications."""
while True:
notification = await self._notification_queue.get()
if notification.method == "notifications/resources/updated":
yield ResourceNotification(uri=notification.params["uri"])
async def _handle_notification(self, notification: dict) -> None:
"""Handle incoming notifications from server."""
await self._notification_queue.put(notification)
```
### Resource Configuration
In `resources/resource.py`:
```python
@dataclass
class ResourceConfig:
subscribable: bool = False
# ... other config
class Resource:
def __init__(
self,
uri: str,
*,
subscribe: bool = False,
config: ResourceConfig | None = None,
**kwargs
):
if config is None:
config = ResourceConfig()
if subscribe:
config.subscribable = True
self.subscribable = config.subscribable
```
## Edge Cases
1. **Session cleanup** - When a session ends, automatically unsubscribe all its subscriptions.
2. **Non-subscribable resources** - Return error if client tries to subscribe to a resource that doesn't support it.
3. **Resource doesn't exist** - Return error if subscribing to non-existent resource.
4. **Duplicate subscriptions** - Allow same session to subscribe multiple times (idempotent).
5. **stdio transport** - Notifications work over stdio using JSON-RPC 2.0 notification format.
6. **HTTP transport** - For SSE, notifications are sent as events. For streamable HTTP, they're sent in the response stream.
7. **Notification delivery failure** - If session is gone when notification fires, silently skip (already cleaned up).
8. **Resource URI patterns** - Template URIs can be subscribable. Notify on exact URI match only.
## Testing
Add `tests/server/test_subscriptions.py`:
```python
async def test_subscribe_resource():
mcp = FastMCP("test")
@mcp.resource("data://metrics", subscribe=True)
def metrics():
return {"value": 42}
async with Client(mcp) as client:
await client.subscribe_resource("data://metrics")
assert "data://metrics" in client.list_subscriptions()
async def test_notification_delivery():
mcp = FastMCP("test")
@mcp.resource("data://metrics", subscribe=True)
def metrics():
return {"value": current_value}
@mcp.tool
async def update(ctx: Context, value: int):
global current_value
current_value = value
await ctx.notify_resource_changed("data://metrics")
async with Client(mcp) as client:
await client.subscribe_resource("data://metrics")
# Trigger notification
await client.call_tool("update", {"value": 100})
# Receive notification
notification = await asyncio.wait_for(
client.resource_notifications().__anext__(),
timeout=1.0
)
assert notification.uri == "data://metrics"
async def test_non_subscribable_resource():
mcp = FastMCP("test")
@mcp.resource("data://config") # No subscribe=True
def config():
return {"key": "value"}
async with Client(mcp) as client:
with pytest.raises(McpError, match="not subscribable"):
await client.subscribe_resource("data://config")
async def test_session_cleanup():
mcp = FastMCP("test")
@mcp.resource("data://metrics", subscribe=True)
def metrics():
return {}
async with Client(mcp) as client1:
await client1.subscribe_resource("data://metrics")
# Session ended, subscriptions should be cleaned up
assert len(mcp._subscription_manager._subscriptions["data://metrics"]) == 0
```
## Documentation
Add to `docs/servers/resources.mdx`:
- Resource subscriptions overview
- Declaring subscribable resources
- Notifying subscribers
- Client subscription API
- Real-time updates pattern
- Comparison with polling
Add example in `docs/examples/`:
```python
# examples/subscriptions/server.py
"""Real-time metrics server with subscriptions."""
import asyncio
from fastmcp import FastMCP, Context
mcp = FastMCP("Metrics Server")
current_metrics = {"cpu": 0.0, "memory": 0.0}
@mcp.resource("data://metrics", subscribe=True)
def get_metrics() -> dict:
return current_metrics
async def update_metrics_loop(server: FastMCP):
"""Background task that updates metrics every 5 seconds."""
while True:
await asyncio.sleep(5)
# Simulate metrics update
current_metrics["cpu"] = random.uniform(0, 100)
current_metrics["memory"] = random.uniform(0, 100)
# Notify all subscribers
await server.notify_resource_changed("data://metrics")
@mcp.lifespan
async def lifespan(server):
task = asyncio.create_task(update_metrics_loop(server))
yield
task.cancel()
```
```python
# examples/subscriptions/client.py
"""Client that subscribes to real-time metrics."""
import asyncio
from fastmcp import Client
async def main():
async with Client("http://localhost:8000/mcp") as client:
# Subscribe
await client.subscribe_resource("data://metrics")
print("Subscribed to metrics")
# Listen for updates
async for notification in client.resource_notifications():
data = await client.read_resource(notification.uri)
print(f"Metrics updated: {data}")
if __name__ == "__main__":
asyncio.run(main())
```

542
plans/05-curator-provider.md Normal file
View file

@ -0,0 +1,542 @@
# CuratorProvider (Search Tools)
## Problem
Servers with many tools overwhelm agent context windows:
- A server with 100 tools uses ~50KB+ just listing them
- Agents waste tokens on irrelevant tools
- Response quality degrades with too many options
- Users can't easily scale their servers
Current workarounds:
- Manual tool grouping with visibility/namespacing (requires upfront design)
- Client-side filtering (not all clients support this)
- Multiple specialized servers (operational complexity)
## Solution
A meta-provider that indexes tools and exposes search/discovery functionality. Instead of showing 100 tools, show 1 search tool that returns the relevant subset based on natural language or keywords.
## Modes
### 1. Search Mode (Keyword/Semantic)
Add a `find_tools` tool that searches over your server's tools.
**Keyword mode**: TF-IDF or simple word matching on tool names and descriptions.
**Semantic mode**: Embeddings-based search (requires `sentence-transformers` or OpenAI).
### 2. Hidden Mode (Future)
Hide the underlying tools entirely - clients only see the search tool. This drastically reduces context usage but requires clients to always search first.
## API
### Basic Usage (Keyword Search)
```python
from fastmcp import FastMCP
from fastmcp.server.providers import CuratorProvider
mcp = FastMCP("server")
# Register many tools
@mcp.tool
def search_documents(query: str) -> list[dict]: ...
@mcp.tool
def list_files(directory: str) -> list[str]: ...
@mcp.tool
def delete_file(path: str) -> bool: ...
# ... 97 more tools ...
# Add curator - indexes all tools, adds search
mcp.add_provider(CuratorProvider(
source=mcp._provider,
mode="keyword",
))
```
Now clients see 101 tools: the original 100 + `find_tools`.
### Semantic Search Mode
```python
from fastmcp.server.providers import CuratorProvider
mcp.add_provider(CuratorProvider(
source=mcp._provider,
mode="semantic",
embedding_model="sentence-transformers/all-MiniLM-L6-v2",
))
```
Requires `pip install sentence-transformers` (optional dependency).
### OpenAI Embeddings
```python
mcp.add_provider(CuratorProvider(
source=mcp._provider,
mode="semantic",
embedding_model="openai",
openai_api_key=os.environ["OPENAI_API_KEY"],
))
```
### Hidden Tools Mode (Future)
```python
mcp.add_provider(CuratorProvider(
source=mcp._provider,
mode="semantic",
hide_tools=True, # Only expose find_tools, hide originals
))
```
Clients only see `find_tools`. Must search before calling anything.
## Generated Tool
### find_tools
```python
def find_tools(
query: str,
limit: int = 10,
) -> list[ToolMatch]:
"""Search for tools relevant to your query.
Args:
query: Natural language description of what you want to do
limit: Maximum number of results to return
Returns:
List of matching tools with relevance scores
"""
```
**Example call**:
```python
result = await client.call_tool("find_tools", {
"query": "I need to clean up old files and free disk space",
"limit": 5
})
```
**Example result**:
```json
[
{
"name": "list_files",
"description": "List files in a directory with filters",
"relevance": 0.92,
"reasoning": "Useful for finding files to evaluate"
},
{
"name": "get_file_age",
"description": "Get the last modified time of a file",
"relevance": 0.87,
"reasoning": "Helps identify old files"
},
{
"name": "delete_files",
"description": "Delete files matching a pattern",
"relevance": 0.85,
"reasoning": "Performs the actual cleanup"
},
{
"name": "get_disk_usage",
"description": "Get disk usage statistics",
"relevance": 0.78,
"reasoning": "Verify space was freed"
}
]
```
## Implementation
### Location
- `src/fastmcp/server/providers/curator.py` - Main implementation
- `src/fastmcp/server/providers/curator_keyword.py` - Keyword search
- `src/fastmcp/server/providers/curator_semantic.py` - Semantic search (optional)
### CuratorProvider
```python
from fastmcp.server.providers import Provider
from fastmcp.tools import Tool
class CuratorProvider(Provider):
"""Meta-provider that adds tool search capabilities."""
def __init__(
self,
source: Provider,
mode: Literal["keyword", "semantic"] = "keyword",
hide_tools: bool = False,
embedding_model: str | None = None,
openai_api_key: str | None = None,
):
self.source = source
self.mode = mode
self.hide_tools = hide_tools
# Build search index
if mode == "keyword":
self.searcher = KeywordSearcher()
else:
self.searcher = SemanticSearcher(
model=embedding_model,
openai_api_key=openai_api_key,
)
self._index_built = False
async def list_tools(self) -> Sequence[Tool]:
"""Return source tools + find_tools."""
source_tools = await self.source.list_tools()
if not self._index_built:
await self._build_index(source_tools)
self._index_built = True
# Add find_tools
find_tool = self._create_find_tools_tool()
if self.hide_tools:
return [find_tool]
else:
return [*source_tools, find_tool]
async def get_tool(self, name: str) -> Tool | None:
if name == "find_tools":
return self._create_find_tools_tool()
if self.hide_tools:
return None
return await self.source.get_tool(name)
async def _build_index(self, tools: Sequence[Tool]) -> None:
"""Build search index from tools."""
for tool in tools:
self.searcher.add(
name=tool.name,
description=tool.description or "",
tags=" ".join(tool.tags) if tool.tags else "",
)
def _create_find_tools_tool(self) -> Tool:
"""Create the find_tools tool."""
async def find_tools_handler(query: str, limit: int = 10) -> list[dict]:
results = await self.searcher.search(query, limit=limit)
# Fetch full tool info for each result
tools = []
for match in results:
tool = await self.source.get_tool(match.name)
if tool:
tools.append({
"name": tool.name,
"description": tool.description,
"relevance": match.score,
"reasoning": match.reasoning,
})
return tools
return Tool.from_function(
find_tools_handler,
name="find_tools",
description="Search for tools relevant to your query",
)
```
### KeywordSearcher
```python
from collections import Counter
import math
class KeywordSearcher:
"""TF-IDF based keyword search."""
def __init__(self):
self.documents: dict[str, str] = {} # name -> text
self.vocab: set[str] = set()
self.idf: dict[str, float] = {}
def add(self, name: str, description: str, tags: str = "") -> None:
"""Add a tool to the index."""
text = f"{name} {description} {tags}".lower()
self.documents[name] = text
self.vocab.update(self._tokenize(text))
async def search(self, query: str, limit: int = 10) -> list[ToolMatch]:
"""Search for relevant tools."""
# Build IDF if not built
if not self.idf:
self._build_idf()
query_tokens = self._tokenize(query.lower())
scores = {}
for name, doc_text in self.documents.items():
doc_tokens = self._tokenize(doc_text)
score = self._compute_score(query_tokens, doc_tokens)
scores[name] = score
# Sort by score descending
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [
ToolMatch(
name=name,
score=score,
reasoning=self._generate_reasoning(name, query),
)
for name, score in ranked[:limit]
if score > 0
]
def _tokenize(self, text: str) -> list[str]:
"""Simple tokenization."""
return text.split()
def _build_idf(self) -> None:
"""Build IDF scores for vocabulary."""
n_docs = len(self.documents)
for word in self.vocab:
df = sum(1 for doc in self.documents.values() if word in doc)
self.idf[word] = math.log(n_docs / (1 + df))
def _compute_score(self, query_tokens: list[str], doc_tokens: list[str]) -> float:
"""Compute TF-IDF similarity."""
doc_counts = Counter(doc_tokens)
score = 0.0
for token in query_tokens:
if token in doc_counts:
tf = doc_counts[token] / len(doc_tokens)
idf = self.idf.get(token, 0)
score += tf * idf
return score
def _generate_reasoning(self, name: str, query: str) -> str:
"""Generate simple reasoning for why tool matched."""
# Find common words
query_words = set(self._tokenize(query.lower()))
doc_words = set(self._tokenize(self.documents[name]))
common = query_words & doc_words
if common:
return f"Matches keywords: {', '.join(sorted(common)[:3])}"
return "Relevant based on semantic similarity"
```
### SemanticSearcher
```python
from sentence_transformers import SentenceTransformer
import numpy as np
class SemanticSearcher:
"""Embeddings-based semantic search."""
def __init__(
self,
model: str = "sentence-transformers/all-MiniLM-L6-v2",
openai_api_key: str | None = None,
):
if model == "openai":
self.use_openai = True
self.openai_api_key = openai_api_key
else:
self.use_openai = False
self.model = SentenceTransformer(model)
self.documents: dict[str, str] = {}
self.embeddings: dict[str, np.ndarray] = {}
def add(self, name: str, description: str, tags: str = "") -> None:
"""Add a tool to the index."""
text = f"{name}. {description}"
self.documents[name] = text
# Embed immediately (could defer to first search)
if self.use_openai:
embedding = self._embed_openai(text)
else:
embedding = self.model.encode(text)
self.embeddings[name] = embedding
async def search(self, query: str, limit: int = 10) -> list[ToolMatch]:
"""Search for relevant tools using embeddings."""
# Embed query
if self.use_openai:
query_embedding = self._embed_openai(query)
else:
query_embedding = self.model.encode(query)
# Compute cosine similarity
scores = {}
for name, doc_embedding in self.embeddings.items():
similarity = np.dot(query_embedding, doc_embedding) / (
np.linalg.norm(query_embedding) * np.linalg.norm(doc_embedding)
)
scores[name] = float(similarity)
# Sort by similarity
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [
ToolMatch(
name=name,
score=score,
reasoning=self._generate_reasoning(name, query, score),
)
for name, score in ranked[:limit]
if score > 0.3 # Threshold
]
def _embed_openai(self, text: str) -> np.ndarray:
"""Get embedding from OpenAI."""
import openai
client = openai.OpenAI(api_key=self.openai_api_key)
response = client.embeddings.create(
input=text,
model="text-embedding-3-small"
)
return np.array(response.data[0].embedding)
def _generate_reasoning(self, name: str, query: str, score: float) -> str:
if score > 0.8:
return "Highly relevant to your request"
elif score > 0.6:
return "Likely useful for this task"
else:
return "May be relevant"
```
### ToolMatch
```python
from dataclasses import dataclass
@dataclass
class ToolMatch:
name: str
score: float
reasoning: str
```
## Edge Cases
1. **Empty index** - If source has no tools, find_tools returns empty list.
2. **Index rebuild** - If tools change (hot reload), should we rebuild index? For now, no - index is static after first build.
3. **Large servers** - Embedding 1000 tools at startup could be slow. Consider lazy indexing.
4. **Query quality** - Vague queries ("do something") will return poor results. Document best practices.
5. **Hide tools mode** - If tools are hidden, agents MUST use find_tools first. Calling a tool by name directly fails.
6. **Concurrent searches** - Index is read-only after building, safe for concurrent access.
## Testing
Add `tests/server/providers/test_curator.py`:
```python
async def test_curator_keyword_search():
provider = LocalProvider()
@provider.tool
def search_documents(query: str) -> list:
"""Search through documents."""
return []
@provider.tool
def delete_file(path: str) -> bool:
"""Delete a file."""
return True
curator = CuratorProvider(provider, mode="keyword")
tools = await curator.list_tools()
assert len(tools) == 3 # original 2 + find_tools
# Call find_tools
find_tool = await curator.get_tool("find_tools")
result = await find_tool.fn(query="search for documents", limit=5)
assert len(result) > 0
assert result[0]["name"] == "search_documents"
assert result[0]["relevance"] > 0
async def test_curator_semantic_search():
pytest.importorskip("sentence_transformers")
provider = LocalProvider()
@provider.tool
def find_files(pattern: str) -> list:
"""Locate files matching a pattern."""
return []
@provider.tool
def delete_data(id: str) -> bool:
"""Remove data by ID."""
return True
curator = CuratorProvider(provider, mode="semantic")
find_tool = await curator.get_tool("find_tools")
result = await find_tool.fn(query="search for files", limit=5)
# Semantic search should match "find_files" even though words differ
assert any(r["name"] == "find_files" for r in result)
```
## Documentation
Add to `docs/servers/providers/curator.mdx`:
- Why search over tools matters
- Keyword vs semantic modes
- Setting up embeddings
- Best practices for queries
- Performance considerations
- Comparison with manual grouping
## Dependencies
Make `sentence-transformers` an optional dependency:
```toml
[project.optional-dependencies]
semantic-search = [
"sentence-transformers>=2.0.0",
]
```
For OpenAI mode, require `openai` (already a dependency).
## Future Enhancements
1. **Agent mode** - Full conversational agent instead of just search
2. **Caching** - Cache query results
3. **Multi-modal search** - Search across tools + resources + prompts
4. **Relevance feedback** - Learn from which tools users actually call
5. **Custom scoring** - Allow users to provide scoring functions

628
plans/06-dynamic-resources.md Normal file
View file

@ -0,0 +1,628 @@
# Dynamic Resources
## Problem
Tools that return large results bloat agent context:
- Search results: 10MB of JSON
- Generated files: large CSV, PDF, images
- Log files: thousands of lines
- Database dumps: extensive data
Current workarounds:
- Return truncated data (loses information)
- Return everything (wastes tokens)
- Write to filesystem manually (not portable, cleanup issues)
The GitHub MCP server pioneered a pattern: write large results to a file and return a reference. Clients can then read/search the file as needed.
## Solution
Allow tools to create ephemeral resources at runtime that clients can read, search, and navigate without bloating context.
## API
### Basic Usage
```python
from fastmcp import FastMCP, Context
mcp = FastMCP("server")
@mcp.tool
async def search(query: str, ctx: Context) -> str:
# Perform search - returns 10MB of data
results = do_search(query)
# Create ephemeral resource instead of returning everything
uri = await ctx.create_resource(
f"results://search/{ctx.request_id}",
content=results,
ttl=3600, # Auto-cleanup after 1 hour
)
return f"Found {len(results)} results. Access at {uri}"
```
Client can then:
```python
result = await client.call_tool("search", {"query": "python async"})
# "Found 1000 results. Access at results://search/abc123"
# Read the full results
data = await client.read_resource("results://search/abc123")
# Or use resource tools (if ResourceToolsProvider is enabled)
preview = await client.call_tool("read_resource", {
"uri": "results://search/abc123",
"limit": 10
})
```
### With Metadata
```python
uri = await ctx.create_resource(
"results://analysis/output.json",
content={"data": [...], "summary": "..."},
name="Analysis Results",
description="Detailed analysis output",
mime_type="application/json",
ttl=7200, # 2 hours
)
```
### Binary Content
```python
@mcp.tool
async def generate_pdf(ctx: Context) -> str:
pdf_bytes = create_pdf()
uri = await ctx.create_resource(
f"output://pdf/{ctx.request_id}.pdf",
content=pdf_bytes,
mime_type="application/pdf",
ttl=1800,
)
return f"PDF generated: {uri}"
```
### Update Resource
```python
@mcp.tool
async def update_results(ctx: Context, uri: str, new_data: dict) -> str:
# Update existing resource
await ctx.update_resource(uri, content=new_data)
return f"Updated {uri}"
```
### Delete Resource
```python
@mcp.tool
async def cleanup(ctx: Context, uri: str) -> str:
await ctx.delete_resource(uri)
return f"Deleted {uri}"
```
## Storage Backends
### In-Memory (Default)
```python
mcp = FastMCP("server")
# Uses in-memory storage - lost on restart
```
### Filesystem
```python
from fastmcp.server.dynamic_resources import FilesystemStorage
mcp = FastMCP(
"server",
dynamic_resource_storage=FilesystemStorage(
directory="/tmp/mcp-resources"
)
)
```
### Redis
```python
from fastmcp.server.dynamic_resources import RedisStorage
mcp = FastMCP(
"server",
dynamic_resource_storage=RedisStorage(
url="redis://localhost:6379"
)
)
```
### S3
```python
from fastmcp.server.dynamic_resources import S3Storage
mcp = FastMCP(
"server",
dynamic_resource_storage=S3Storage(
bucket="my-mcp-resources",
prefix="dynamic/",
)
)
```
## Implementation
### Location
- `src/fastmcp/server/dynamic_resources.py` - Core implementation
- `src/fastmcp/server/context.py` - Add context methods
- `src/fastmcp/server/server.py` - Integrate with server
### Storage Interface
```python
from abc import ABC, abstractmethod
from typing import Protocol
class DynamicResourceStorage(Protocol):
"""Storage backend for dynamic resources."""
async def write(
self,
uri: str,
content: str | bytes | dict,
*,
metadata: dict | None = None,
ttl: int | None = None,
) -> None:
"""Write a resource."""
...
async def read(self, uri: str) -> tuple[bytes, dict]:
"""Read a resource. Returns (content, metadata)."""
...
async def delete(self, uri: str) -> None:
"""Delete a resource."""
...
async def exists(self, uri: str) -> bool:
"""Check if resource exists."""
...
async def list(self, prefix: str | None = None) -> list[str]:
"""List all URIs, optionally filtered by prefix."""
...
```
### In-Memory Storage
```python
import asyncio
from datetime import datetime, timedelta
class InMemoryStorage:
"""In-memory storage with TTL support."""
def __init__(self):
self._data: dict[str, tuple[bytes, dict, datetime | None]] = {}
self._cleanup_task: asyncio.Task | None = None
async def write(
self,
uri: str,
content: str | bytes | dict,
*,
metadata: dict | None = None,
ttl: int | None = None,
) -> None:
# Serialize content
if isinstance(content, dict):
content_bytes = json.dumps(content).encode()
elif isinstance(content, str):
content_bytes = content.encode()
else:
content_bytes = content
# Calculate expiry
expiry = None
if ttl:
expiry = datetime.now() + timedelta(seconds=ttl)
self._data[uri] = (content_bytes, metadata or {}, expiry)
# Start cleanup task if not running
if not self._cleanup_task:
self._cleanup_task = asyncio.create_task(self._cleanup_loop())
async def read(self, uri: str) -> tuple[bytes, dict]:
if uri not in self._data:
raise KeyError(f"Resource not found: {uri}")
content, metadata, expiry = self._data[uri]
# Check expiry
if expiry and datetime.now() > expiry:
del self._data[uri]
raise KeyError(f"Resource expired: {uri}")
return content, metadata
async def delete(self, uri: str) -> None:
self._data.pop(uri, None)
async def exists(self, uri: str) -> bool:
return uri in self._data
async def list(self, prefix: str | None = None) -> list[str]:
if prefix:
return [uri for uri in self._data if uri.startswith(prefix)]
return list(self._data.keys())
async def _cleanup_loop(self) -> None:
"""Periodically remove expired resources."""
while True:
await asyncio.sleep(60) # Check every minute
now = datetime.now()
expired = [
uri for uri, (_, _, expiry) in self._data.items()
if expiry and now > expiry
]
for uri in expired:
del self._data[uri]
if not self._data:
break # Stop cleanup if empty
```
### Filesystem Storage
```python
import aiofiles
import os
from pathlib import Path
class FilesystemStorage:
"""Filesystem-backed storage."""
def __init__(self, directory: str | Path):
self.directory = Path(directory)
self.directory.mkdir(parents=True, exist_ok=True)
async def write(
self,
uri: str,
content: str | bytes | dict,
*,
metadata: dict | None = None,
ttl: int | None = None,
) -> None:
# Create path from URI
path = self._uri_to_path(uri)
path.parent.mkdir(parents=True, exist_ok=True)
# Serialize
if isinstance(content, dict):
content_bytes = json.dumps(content).encode()
elif isinstance(content, str):
content_bytes = content.encode()
else:
content_bytes = content
# Write content
async with aiofiles.open(path, "wb") as f:
await f.write(content_bytes)
# Write metadata
meta_path = path.with_suffix(path.suffix + ".meta")
async with aiofiles.open(meta_path, "w") as f:
meta = metadata or {}
if ttl:
meta["expires_at"] = (datetime.now() + timedelta(seconds=ttl)).isoformat()
await f.write(json.dumps(meta))
async def read(self, uri: str) -> tuple[bytes, dict]:
path = self._uri_to_path(uri)
if not path.exists():
raise KeyError(f"Resource not found: {uri}")
# Check expiry
meta_path = path.with_suffix(path.suffix + ".meta")
if meta_path.exists():
async with aiofiles.open(meta_path, "r") as f:
metadata = json.loads(await f.read())
if "expires_at" in metadata:
expiry = datetime.fromisoformat(metadata["expires_at"])
if datetime.now() > expiry:
# Delete expired
path.unlink()
meta_path.unlink()
raise KeyError(f"Resource expired: {uri}")
else:
metadata = {}
# Read content
async with aiofiles.open(path, "rb") as f:
content = await f.read()
return content, metadata
async def delete(self, uri: str) -> None:
path = self._uri_to_path(uri)
path.unlink(missing_ok=True)
path.with_suffix(path.suffix + ".meta").unlink(missing_ok=True)
async def exists(self, uri: str) -> bool:
return self._uri_to_path(uri).exists()
async def list(self, prefix: str | None = None) -> list[str]:
uris = []
for path in self.directory.rglob("*"):
if path.is_file() and not path.name.endswith(".meta"):
uri = self._path_to_uri(path)
if not prefix or uri.startswith(prefix):
uris.append(uri)
return uris
def _uri_to_path(self, uri: str) -> Path:
"""Convert URI to filesystem path."""
# Strip scheme
path_part = uri.split("://", 1)[1] if "://" in uri else uri
return self.directory / path_part
def _path_to_uri(self, path: Path) -> str:
"""Convert filesystem path to URI."""
rel = path.relative_to(self.directory)
return f"file://{rel}"
```
### Context Methods
In `context.py`:
```python
class Context:
async def create_resource(
self,
uri: str,
content: str | bytes | dict,
*,
name: str | None = None,
description: str | None = None,
mime_type: str | None = None,
ttl: int | None = None,
) -> str:
"""Create a dynamic resource.
Args:
uri: Resource URI
content: Resource content (string, bytes, or dict)
name: Human-readable name
description: Resource description
mime_type: MIME type
ttl: Time-to-live in seconds (default: 86400 = 1 day)
Returns:
The URI of the created resource
"""
metadata = {}
if name:
metadata["name"] = name
if description:
metadata["description"] = description
if mime_type:
metadata["mime_type"] = mime_type
ttl = ttl or 86400 # Default 1 day
await self._server._dynamic_resource_storage.write(
uri, content, metadata=metadata, ttl=ttl
)
return uri
async def update_resource(
self,
uri: str,
content: str | bytes | dict,
) -> None:
"""Update an existing dynamic resource."""
# Read existing metadata
_, metadata = await self._server._dynamic_resource_storage.read(uri)
# Write with updated content, preserve metadata
await self._server._dynamic_resource_storage.write(
uri, content, metadata=metadata
)
async def delete_resource(self, uri: str) -> None:
"""Delete a dynamic resource."""
await self._server._dynamic_resource_storage.delete(uri)
```
### Server Integration
In `server.py`:
```python
class FastMCP:
def __init__(
self,
name: str,
*,
dynamic_resource_storage: DynamicResourceStorage | None = None,
**kwargs
):
self._dynamic_resource_storage = (
dynamic_resource_storage or InMemoryStorage()
)
async def get_resource(self, uri: str) -> Resource | None:
"""Get a resource - check dynamic storage first."""
# Check dynamic resources
if await self._dynamic_resource_storage.exists(uri):
content, metadata = await self._dynamic_resource_storage.read(uri)
# Create a dynamic resource wrapper
return DynamicResource(
uri=uri,
content=content,
name=metadata.get("name"),
description=metadata.get("description"),
mime_type=metadata.get("mime_type"),
)
# Fall back to provider resources
return await super().get_resource(uri)
async def list_resources(self) -> list[Resource]:
"""List resources - include dynamic ones."""
# Get provider resources
provider_resources = await super().list_resources()
# Get dynamic resources
dynamic_uris = await self._dynamic_resource_storage.list()
dynamic_resources = []
for uri in dynamic_uris:
_, metadata = await self._dynamic_resource_storage.read(uri)
dynamic_resources.append(
DynamicResource(
uri=uri,
name=metadata.get("name"),
description=metadata.get("description"),
)
)
return [*provider_resources, *dynamic_resources]
```
## Edge Cases
1. **URI collisions** - Dynamic resource URI conflicts with provider resource. Dynamic takes precedence.
2. **TTL expiry** - Resource expires while being read. Return error, client should handle.
3. **Large content** - 100MB+ files in memory. Use filesystem or S3 storage.
4. **Session cleanup** - When session ends, should we delete its resources? Optional, controlled by TTL.
5. **Concurrent writes** - Two tools write to same URI. Last write wins (no locking).
6. **URI schemes** - Any scheme is allowed. Convention: `results://`, `output://`, `temp://`.
7. **Storage backend failure** - If Redis is down, operations fail with clear error.
## Testing
Add `tests/server/test_dynamic_resources.py`:
```python
async def test_create_resource():
mcp = FastMCP("test")
@mcp.tool
async def create(ctx: Context) -> str:
uri = await ctx.create_resource(
"results://test",
content={"data": [1, 2, 3]},
ttl=60
)
return uri
async with Client(mcp) as client:
uri = await client.call_tool("create", {})
assert uri == "results://test"
# Read it back
resource = await client.read_resource(uri)
assert resource.contents[0].text == '{"data": [1, 2, 3]}'
async def test_resource_ttl():
storage = InMemoryStorage()
await storage.write("test://resource", "data", ttl=1)
# Should exist immediately
assert await storage.exists("test://resource")
# Wait for expiry
await asyncio.sleep(1.5)
# Should be gone
with pytest.raises(KeyError):
await storage.read("test://resource")
async def test_filesystem_storage(tmp_path):
storage = FilesystemStorage(tmp_path)
await storage.write(
"output://file.json",
{"key": "value"},
metadata={"name": "Test File"}
)
content, metadata = await storage.read("output://file.json")
assert json.loads(content) == {"key": "value"}
assert metadata["name"] == "Test File"
# Check file exists
assert (tmp_path / "output" / "file.json").exists()
```
## Documentation
Add to `docs/servers/dynamic-resources.mdx`:
- Why dynamic resources matter
- Creating resources from tools
- Storage backends (memory, filesystem, Redis, S3)
- TTL and cleanup
- Best practices (when to use vs returning data)
- Integration with ResourceToolsProvider
Add example in `docs/examples/`:
```python
# examples/dynamic_resources/search_server.py
"""Search server that returns large results as resources."""
from fastmcp import FastMCP, Context
mcp = FastMCP("Search Server")
@mcp.tool
async def search_logs(query: str, ctx: Context) -> str:
"""Search through millions of log lines."""
results = search_engine.search(query) # Returns 10MB
# Instead of returning all 10MB...
uri = await ctx.create_resource(
f"results://search/{ctx.request_id}",
content=results,
name=f"Search Results: {query}",
description=f"Found {len(results)} matches for '{query}'",
mime_type="application/json",
ttl=3600,
)
return f"Search complete. Found {len(results)} results. Access at {uri}"
```
## Future Enhancements
1. **Resource pagination** - Auto-paginate large dynamic resources
2. **Compression** - Compress content before storing
3. **Access control** - Per-resource auth checks
4. **Versioning** - Keep multiple versions of same URI
5. **Search** - Full-text search over dynamic resources

2
src/fastmcp/server/transforms/__init__.py
View file

@ -222,6 +222,7 @@ class Transform:
# Re-export built-in transforms (must be after Transform class to avoid circular imports)
from fastmcp.server.transforms.enabled import Enabled, is_enabled # noqa: E402
from fastmcp.server.transforms.namespace import Namespace # noqa: E402
from fastmcp.server.transforms.prompts_as_tools import PromptsAsTools # noqa: E402
from fastmcp.server.transforms.resources_as_tools import ResourcesAsTools # noqa: E402
from fastmcp.server.transforms.tool_transform import ToolTransform # noqa: E402
from fastmcp.server.transforms.version_filter import VersionFilter # noqa: E402
@ -233,6 +234,7 @@ __all__ = [
"GetResourceTemplateNext",
"GetToolNext",
"Namespace",
"PromptsAsTools",
"ResourcesAsTools",
"ToolTransform",
"Transform",

175
src/fastmcp/server/transforms/prompts_as_tools.py Normal file
View file

@ -0,0 +1,175 @@
"""Transform that exposes prompts as tools.
This transform generates tools for listing and getting prompts, enabling
clients that only support tools to access prompt functionality.
Example:
```python
from fastmcp import FastMCP
from fastmcp.server.transforms import PromptsAsTools
mcp = FastMCP("Server")
mcp.add_transform(PromptsAsTools(mcp))
# Now has list_prompts and get_prompt tools
```
"""
from __future__ import annotations
import json
from collections.abc import Sequence
from typing import TYPE_CHECKING, Annotated, Any
from mcp.types import TextContent
from fastmcp.server.transforms import GetToolNext, Transform
from fastmcp.tools.tool import Tool
from fastmcp.utilities.versions import VersionSpec
if TYPE_CHECKING:
from fastmcp.server.providers.base import Provider
# Note: FastMCP imported inside tools to avoid circular import
class PromptsAsTools(Transform):
"""Transform that adds tools for listing and getting prompts.
Generates two tools:
- `list_prompts`: Lists all prompts from the provider
- `get_prompt`: Gets a specific prompt with optional arguments
The transform captures a provider reference at construction and queries it
for prompts when the generated tools are called. When used with FastMCP,
the provider's auth and visibility filtering is automatically applied.
Example:
```python
mcp = FastMCP("Server")
mcp.add_transform(PromptsAsTools(mcp))
# Now has list_prompts and get_prompt tools
```
"""
def __init__(self, provider: Provider) -> None:
"""Initialize the transform with a provider reference.
Args:
provider: The provider to query for prompts. Typically this is
the same FastMCP server the transform is added to.
"""
self._provider = provider
def __repr__(self) -> str:
return f"PromptsAsTools({self._provider!r})"
async def list_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]:
"""Add prompt tools to the tool list."""
return [
*tools,
self._make_list_prompts_tool(),
self._make_get_prompt_tool(),
]
async def get_tool(
self, name: str, call_next: GetToolNext, *, version: VersionSpec | None = None
) -> Tool | None:
"""Get a tool by name, including generated prompt tools."""
# Check if it's one of our generated tools
if name == "list_prompts":
return self._make_list_prompts_tool()
if name == "get_prompt":
return self._make_get_prompt_tool()
# Otherwise delegate to downstream
return await call_next(name, version=version)
def _make_list_prompts_tool(self) -> Tool:
"""Create the list_prompts tool."""
provider = self._provider
async def list_prompts() -> str:
"""List all available prompts.
Returns JSON with prompt metadata including name, description,
and optional arguments.
"""
prompts = await provider.list_prompts()
result: list[dict[str, Any]] = []
for p in prompts:
result.append(
{
"name": p.name,
"description": p.description,
"arguments": [
{
"name": arg.name,
"description": arg.description,
"required": arg.required,
}
for arg in (p.arguments or [])
],
}
)
return json.dumps(result, indent=2)
return Tool.from_function(fn=list_prompts)
def _make_get_prompt_tool(self) -> Tool:
"""Create the get_prompt tool."""
provider = self._provider
async def get_prompt(
name: Annotated[str, "The name of the prompt to get"],
arguments: Annotated[
dict[str, Any] | None,
"Optional arguments for the prompt",
] = None,
) -> str:
"""Get a prompt by name with optional arguments.
Returns the rendered prompt as JSON with a messages array.
Arguments should be provided as a dict mapping argument names to values.
"""
from fastmcp.server.server import FastMCP
# Use FastMCP.render_prompt() if available - runs middleware chain
if isinstance(provider, FastMCP):
result = await provider.render_prompt(name, arguments=arguments or {})
return _format_prompt_result(result)
# Fallback for plain providers - no middleware
prompt = await provider.get_prompt(name)
if prompt is None:
raise ValueError(f"Prompt not found: {name}")
result = await prompt._render(arguments or {})
return _format_prompt_result(result)
return Tool.from_function(fn=get_prompt)
def _format_prompt_result(result: Any) -> str:
"""Format PromptResult for tool output.
Returns JSON with the messages array. Preserves embedded resources
as structured JSON objects.
"""
messages = []
for msg in result.messages:
if isinstance(msg.content, TextContent):
content = msg.content.text
else:
# Preserve structured content (e.g., EmbeddedResource) as dict
content = msg.content.model_dump(mode="json", exclude_none=True)
messages.append(
{
"role": msg.role,
"content": content,
}
)
return json.dumps({"messages": messages}, indent=2)

210
tests/server/transforms/test_prompts_as_tools.py Normal file
View file

@ -0,0 +1,210 @@
"""Tests for PromptsAsTools transform."""
import json
import pytest
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.server.transforms import PromptsAsTools
class TestPromptsAsToolsBasic:
"""Test basic PromptsAsTools functionality."""
async def test_adds_list_prompts_tool(self):
"""Transform adds list_prompts tool."""
mcp = FastMCP("Test")
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
tools = await client.list_tools()
tool_names = {t.name for t in tools}
assert "list_prompts" in tool_names
async def test_adds_get_prompt_tool(self):
"""Transform adds get_prompt tool."""
mcp = FastMCP("Test")
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
tools = await client.list_tools()
tool_names = {t.name for t in tools}
assert "get_prompt" in tool_names
async def test_preserves_existing_tools(self):
"""Transform preserves existing tools."""
mcp = FastMCP("Test")
@mcp.tool
def my_tool() -> str:
return "result"
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
tools = await client.list_tools()
tool_names = {t.name for t in tools}
assert "my_tool" in tool_names
assert "list_prompts" in tool_names
assert "get_prompt" in tool_names
class TestListPromptsTool:
"""Test the list_prompts tool."""
async def test_lists_prompts(self):
"""list_prompts returns prompt metadata."""
mcp = FastMCP("Test")
@mcp.prompt
def analyze_code() -> str:
"""Analyze code for issues."""
return "Analyze this code"
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
result = await client.call_tool("list_prompts", {})
prompts = json.loads(result.data)
assert len(prompts) == 1
assert prompts[0]["name"] == "analyze_code"
assert prompts[0]["description"] == "Analyze code for issues."
async def test_lists_prompt_with_arguments(self):
"""list_prompts includes argument metadata."""
mcp = FastMCP("Test")
@mcp.prompt
def analyze_code(code: str, language: str = "python") -> str:
"""Analyze code for issues."""
return f"Analyze this {language} code:\n{code}"
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
result = await client.call_tool("list_prompts", {})
prompts = json.loads(result.data)
assert len(prompts) == 1
args = prompts[0]["arguments"]
assert len(args) == 2
# Check required arg
code_arg = next(a for a in args if a["name"] == "code")
assert code_arg["required"] is True
# Check optional arg
lang_arg = next(a for a in args if a["name"] == "language")
assert lang_arg["required"] is False
async def test_empty_when_no_prompts(self):
"""list_prompts returns empty list when no prompts exist."""
mcp = FastMCP("Test")
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
result = await client.call_tool("list_prompts", {})
assert json.loads(result.data) == []
class TestGetPromptTool:
"""Test the get_prompt tool."""
async def test_gets_prompt_without_arguments(self):
"""get_prompt gets a prompt with no arguments."""
mcp = FastMCP("Test")
@mcp.prompt
def simple_prompt() -> str:
"""A simple prompt."""
return "Hello, world!"
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
result = await client.call_tool("get_prompt", {"name": "simple_prompt"})
response = json.loads(result.data)
assert "messages" in response
assert len(response["messages"]) == 1
assert response["messages"][0]["role"] == "user"
assert "Hello, world!" in response["messages"][0]["content"]
async def test_gets_prompt_with_arguments(self):
"""get_prompt gets a prompt with arguments."""
mcp = FastMCP("Test")
@mcp.prompt
def analyze_code(code: str, language: str = "python") -> str:
"""Analyze code."""
return f"Analyze this {language} code:\n{code}"
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
result = await client.call_tool(
"get_prompt",
{
"name": "analyze_code",
"arguments": {"code": "x = 1", "language": "python"},
},
)
response = json.loads(result.data)
assert "messages" in response
content = response["messages"][0]["content"]
assert "python" in content
assert "x = 1" in content
async def test_error_on_unknown_prompt(self):
"""get_prompt raises error for unknown prompt name."""
from fastmcp.exceptions import ToolError
mcp = FastMCP("Test")
mcp.add_transform(PromptsAsTools(mcp))
async with Client(mcp) as client:
with pytest.raises(ToolError, match="Unknown prompt"):
await client.call_tool("get_prompt", {"name": "unknown_prompt"})
class TestPromptsAsToolsWithNamespace:
"""Test PromptsAsTools combined with other transforms."""
async def test_works_with_namespace_on_provider(self):
"""PromptsAsTools works when provider has Namespace transform."""
from fastmcp.server.providers import FastMCPProvider
from fastmcp.server.transforms import Namespace
sub = FastMCP("Sub")
@sub.prompt
def my_prompt() -> str:
"""A prompt."""
return "Hello"
main = FastMCP("Main")
provider = FastMCPProvider(sub)
provider.add_transform(Namespace("sub"))
main.add_provider(provider)
main.add_transform(PromptsAsTools(main))
async with Client(main) as client:
result = await client.call_tool("list_prompts", {})
prompts = json.loads(result.data)
# Prompt should have namespaced name
assert len(prompts) == 1
assert prompts[0]["name"] == "sub_my_prompt"
class TestPromptsAsToolsRepr:
"""Test PromptsAsTools repr."""
def test_repr(self):
"""Transform has useful repr."""
mcp = FastMCP("Test")
transform = PromptsAsTools(mcp)
assert "PromptsAsTools" in repr(transform)