diff --git a/docs/development/v3-notes/v3-features.mdx b/docs/development/v3-notes/v3-features.mdx index 5217011a2..4f02e705a 100644 --- a/docs/development/v3-notes/v3-features.mdx +++ b/docs/development/v3-notes/v3-features.mdx @@ -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 diff --git a/docs/docs.json b/docs/docs.json index 51ab4a5c7..8e776679a 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -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", diff --git a/docs/servers/providers/prompts-as-tools.mdx b/docs/servers/providers/prompts-as-tools.mdx new file mode 100644 index 000000000..d7c504fc7 --- /dev/null +++ b/docs/servers/providers/prompts-as-tools.mdx @@ -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' + + + +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. diff --git a/examples/prompts_as_tools/client.py b/examples/prompts_as_tools/client.py new file mode 100644 index 000000000..43f9bb477 --- /dev/null +++ b/examples/prompts_as_tools/client.py @@ -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()) diff --git a/examples/prompts_as_tools/server.py b/examples/prompts_as_tools/server.py new file mode 100644 index 000000000..912ef0331 --- /dev/null +++ b/examples/prompts_as_tools/server.py @@ -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() diff --git a/plans/03-health-endpoint.md b/plans/03-health-endpoint.md new file mode 100644 index 000000000..d7bbdc86c --- /dev/null +++ b/plans/03-health-endpoint.md @@ -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 diff --git a/plans/04-resource-subscriptions.md b/plans/04-resource-subscriptions.md new file mode 100644 index 000000000..a82cb8ad5 --- /dev/null +++ b/plans/04-resource-subscriptions.md @@ -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()) +``` diff --git a/plans/05-curator-provider.md b/plans/05-curator-provider.md new file mode 100644 index 000000000..6054db8cf --- /dev/null +++ b/plans/05-curator-provider.md @@ -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 diff --git a/plans/06-dynamic-resources.md b/plans/06-dynamic-resources.md new file mode 100644 index 000000000..391187333 --- /dev/null +++ b/plans/06-dynamic-resources.md @@ -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 diff --git a/src/fastmcp/server/transforms/__init__.py b/src/fastmcp/server/transforms/__init__.py index b68b2f4b6..63225854a 100644 --- a/src/fastmcp/server/transforms/__init__.py +++ b/src/fastmcp/server/transforms/__init__.py @@ -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", diff --git a/src/fastmcp/server/transforms/prompts_as_tools.py b/src/fastmcp/server/transforms/prompts_as_tools.py new file mode 100644 index 000000000..83ef30815 --- /dev/null +++ b/src/fastmcp/server/transforms/prompts_as_tools.py @@ -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) diff --git a/tests/server/transforms/test_prompts_as_tools.py b/tests/server/transforms/test_prompts_as_tools.py new file mode 100644 index 000000000..1583d3b11 --- /dev/null +++ b/tests/server/transforms/test_prompts_as_tools.py @@ -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)