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

Add pattern docs

Browse source
This commit is contained in:
Jeremiah Lowin 2025-04-13 14:16:28 -04:00
commit 9f209357df
5 changed files with 591 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

11
docs/docs.json
View file

@ -10,7 +10,7 @@
"colors": {
"dark": "#f72585",
"light": "#4cc9f0",
"primary": "#3f37c9"
"primary": "#2d00f7"
},
"description": "The fast, Pythonic way to build MCP servers.",
"footer": {
@ -54,6 +54,15 @@
"clients/transports"
]
},
{
"group": "Advanced Patterns",
"pages": [
"patterns/proxying",
"patterns/composition",
"patterns/openapi",
"patterns/fastapi"
]
},
{
"group": "Deployment",
"pages": []

186
docs/patterns/composition.mdx Normal file
View file

@ -0,0 +1,186 @@
---
title: Server Composition
sidebarTitle: Composition
description: Combine multiple FastMCP servers into a single, larger application using mounting.
icon: puzzle-piece
---
As your MCP applications grow, you might want to organize your tools, resources, and prompts into logical modules or reuse existing server components. FastMCP supports composition through the `server.mount()` method, allowing you to combine multiple `FastMCP` instances into a single, unified server.
## Why Compose Servers?
- **Modularity**: Break down large applications into smaller, focused servers (e.g., a `WeatherServer`, a `DatabaseServer`, a `CalendarServer`).
- **Reusability**: Create common utility servers (e.g., a `TextProcessingServer`) and mount them wherever needed.
- **Teamwork**: Different teams can work on separate FastMCP servers that are later combined.
- **Organization**: Keep related functionality grouped together logically.
## Mounting Subservers
The `mount()` method attaches all components (tools, resources, templates, prompts) from one `FastMCP` instance (the *subserver*) onto another (the *main server*). A `prefix` is added to avoid naming conflicts.
```python
from fastmcp import FastMCP
from typing import dict, list
# --- Define Subservers ---
# Weather Service
weather_mcp = FastMCP(name="WeatherService")
@weather_mcp.tool()
def get_forecast(city: str) -> dict:
"""Get weather forecast."""
return {"city": city, "forecast": "Sunny"}
@weather_mcp.resource("data://cities/supported")
def list_supported_cities() -> list[str]:
"""List cities with weather support."""
return ["London", "Paris", "Tokyo"]
# Calculator Service
calc_mcp = FastMCP(name="CalculatorService")
@calc_mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
@calc_mcp.prompt()
def explain_addition() -> str:
"""Explain the concept of addition."""
return "Addition is the process of combining two or more numbers."
# --- Define Main Server ---
main_mcp = FastMCP(name="MainApp")
# --- Mount Subservers ---
# Mount weather service with prefix "weather"
main_mcp.mount("weather", weather_mcp)
# Mount calculator service with prefix "calc"
main_mcp.mount("calc", calc_mcp)
# --- Now, main_mcp contains combined components ---
# Tools:
# - "weather_get_forecast"
# - "calc_add"
# Resources:
# - "weather+data://cities/supported" (prefixed URI)
# Prompts:
# - "calc_explain_addition"
if __name__ == "__main__":
# Run the main server, which now includes components from both subservers
main_mcp.run()
```
### How Mounting Works
When you call `main_mcp.mount(prefix, subserver)`:
1. **Tools**: All tools from `subserver` are added to `main_mcp`. Their names are automatically prefixed using the `prefix` and a default separator (`_`).
- `subserver.tool(name="my_tool")` becomes `main_mcp.tool(name="{prefix}_my_tool")`.
2. **Resources**: All resources from `subserver` are added. Their URIs are prefixed using the `prefix` and a default separator (`+`).
- `subserver.resource(uri="data://info")` becomes `main_mcp.resource(uri="{prefix}+data://info")`.
3. **Resource Templates**: All templates from `subserver` are added. Their URI *templates* are prefixed similarly to resources.
- `subserver.resource(uri="data://{id}")` becomes `main_mcp.resource(uri="{prefix}+data://{id}")`.
4. **Prompts**: All prompts from `subserver` are added, with names prefixed like tools.
- `subserver.prompt(name="my_prompt")` becomes `main_mcp.prompt(name="{prefix}_my_prompt")`.
5. **Lifespan Management**: If the `subserver` has a `lifespan` function defined, it will be automatically executed within the `main_mcp`'s lifespan context. This ensures that setup and teardown logic for the subserver runs correctly.
### Customizing Separators
You might prefer different separators for the prefixed names and URIs. You can customize these when calling `mount()`:
```python
main_mcp.mount(
prefix="api",
app=some_subserver,
tool_separator="/", # Tool name becomes: "api/sub_tool_name"
resource_separator=":", # Resource URI becomes: "api:data://sub_resource"
prompt_separator="." # Prompt name becomes: "api.sub_prompt_name"
)
```
<Warning>
Be cautious when choosing separators. Some MCP clients (like Claude Desktop) might have restrictions on characters allowed in tool names (e.g., `/` might not be supported). The defaults (`_` for names, `+` for URIs) are generally safe.
</Warning>
## Example: Modular Application
```python
# modules/text_utils.py
from fastmcp import FastMCP
from typing import list
text_mcp = FastMCP(name="TextUtilities")
@text_mcp.tool()
def count_words(text: str) -> int:
"""Counts words in a text."""
return len(text.split())
@text_mcp.resource("resource://stopwords")
def get_stopwords() -> list[str]:
"""Return a list of common stopwords."""
return ["the", "a", "is", "in"]
# ------------------------------
# modules/data_api.py
from fastmcp import FastMCP
import random
from typing import dict
data_mcp = FastMCP(name="DataAPI")
@data_mcp.tool()
def fetch_record(record_id: int) -> dict:
"""Fetches a dummy data record."""
return {"id": record_id, "value": random.random()}
@data_mcp.resource("data://schema/{table}")
def get_table_schema(table: str) -> dict:
"""Provides a dummy schema for a table."""
return {"table": table, "columns": ["id", "value"]}
# ------------------------------
# main_app.py
from fastmcp import FastMCP
from modules.text_utils import text_mcp # Import server instances
from modules.data_api import data_mcp
app = FastMCP(name="MainApplication")
# Mount the utility servers
app.mount("text", text_mcp)
app.mount("data", data_mcp)
@app.tool()
def process_and_analyze(record_id: int) -> str:
"""Fetches a record and analyzes its string representation."""
# In a real application, you'd use proper methods to interact between
# mounted tools rather than accessing internal managers
# Get record data
record = {"id": record_id, "value": random.random()}
# Count words in the record string representation
word_count = len(str(record).split())
return (
f"Record {record_id} has {word_count} words in its string "
f"representation."
)
if __name__ == "__main__":
app.run()
```
Now, running `main_app.py` starts a server that exposes:
- `text_count_words`
- `data_fetch_record`
- `process_and_analyze`
- `text+resource://stopwords`
- `data+data://schema/{table}` (template)
This pattern promotes code organization and reuse within your FastMCP projects.

114
docs/patterns/fastapi.mdx Normal file
View file

@ -0,0 +1,114 @@
---
title: FastAPI Integration
sidebarTitle: FastAPI
description: Automatically create FastMCP servers directly from FastAPI applications.
icon: square-bolt
---
If you build your APIs using the popular [FastAPI](https://fastapi.tiangolo.com/) framework, FastMCP offers a seamless way to expose your FastAPI application as an MCP server. This leverages the OpenAPI integration internally but simplifies the setup significantly.
## The Goal: FastAPI App -> MCP Server
FastAPI automatically generates an OpenAPI specification for your application. FastMCP uses this built-in capability to create an MCP server that mirrors your API routes.
- FastAPI path operations (`@app.get`, `@app.post`, etc.) become MCP tools, resources, or templates.
- Pydantic models used in FastAPI for request/response validation are used to generate MCP schemas.
- Communication happens directly in memory, making it very efficient.
## Creating from FastAPI App
Use the `FastMCP.from_fastapi()` class method. You only need your FastAPI `app` instance.
```python
import asyncio
from fastapi import FastAPI
from pydantic import BaseModel
from fastmcp import FastMCP, Client # Import FastMCP and Client
# 1. Define your FastAPI application
api_app = FastAPI(title="MyFastAPIApp")
class Item(BaseModel):
name: str
price: float
is_offer: bool | None = None
@api_app.get("/")
def read_root():
return {"Hello": "World"}
@api_app.get("/items/{item_id}") # -> Resource Template
def read_item(item_id: int, q: str | None = None):
# This will become resource://openapi/read_item_items__item_id__get/{item_id}
return {"item_id": item_id, "q": q, "description": f"Details for item {item_id}"}
@api_app.post("/items/") # -> Tool
def create_item(item: Item):
# This will become the 'create_item_items__post' tool
print(f"Creating item: {item.name}")
return {"item_name": item.name, "status": "created"}
# 2. Create the FastMCP server directly from the FastAPI app
# This is an async class method
async def create_mcp_server_from_fastapi():
mcp_server = await FastMCP.from_fastapi(
app=api_app,
name="FastAPI_MCP_Bridge" # Optional name for the MCP server
)
return mcp_server
# 3. (Example) Run the MCP server and test with an in-memory client
async def run_and_test():
server = await create_mcp_server_from_fastapi()
print(f"Created MCP server '{server.name}' from FastAPI app '{api_app.title}'")
# List discovered components
tools = await server.list_tools()
templates = await server.list_resource_templates()
print("Discovered Tools:", [t.name for t in tools])
print("Discovered Templates:", [t.uriTemplate for t in templates])
# Test using an in-memory client
client = Client(server) # Uses FastMCPTransport
async with client:
# Call the tool derived from POST /items/
create_result = await client.call_tool(
"create_item_items__post",
{"name": "MCP Special", "price": 99.99} # Pydantic model fields become args
)
print("Create Item Tool Result:", create_result[0].text) # JSON string
# Read the resource derived from GET /items/{item_id}
read_result = await client.read_resource(
"resource://openapi/read_item_items__item_id__get/42" # Match template URI
)
print("Read Item Resource Result:", read_result[0].text) # JSON string
# In a real scenario, you might run the MCP server via stdio or sse
# print("Running MCP server via stdio...")
# server.run()
if __name__ == "__main__":
# Requires fastapi, uvicorn, httpx:
# uv pip install "fastapi[all]" httpx
try:
asyncio.run(run_and_test())
except ImportError as e:
print(f"Error: {e}. Please install required packages: uv pip install \"fastapi[all]\" httpx")
# Example Output might include:
# Created MCP server 'FastAPI_MCP_Bridge' from FastAPI app 'MyFastAPIApp'
# Discovered Tools: ['read_root___get', 'create_item_items__post']
# Discovered Templates: ['resource://openapi/read_item_items__item_id__get/{item_id}']
# Create Item Tool Result: {"item_name": "MCP Special", "status": "created"}
# Read Item Resource Result: {"item_id": 42, "q": null, "description": "Details for item 42"}
```
### How it Works Internally
1. **OpenAPI Generation**: `from_fastapi` asks the FastAPI `app` for its OpenAPI schema dictionary (`app.openapi()`).
2. **In-Memory Client**: It creates an `httpx.AsyncClient` configured with an `ASGITransport`. This special transport allows `httpx` to call the FastAPI application directly in memory without needing a running web server process.
3. **OpenAPI Integration**: It calls `FastMCP.from_openapi()`, passing the generated schema and the in-memory `httpx` client.
4. **MCP Server Creation**: The standard OpenAPI integration logic then proceeds to parse the schema and create the `Tool`, `Resource`, and `ResourceTemplate` components that wrap calls to the in-memory FastAPI app.
This provides a highly efficient way to expose your FastAPI logic through the MCP protocol, leveraging FastAPI's routing, dependency injection, and validation features.

172
docs/patterns/openapi.mdx Normal file
View file

@ -0,0 +1,172 @@
---
title: OpenAPI Integration
sidebarTitle: OpenAPI
description: Automatically create FastMCP servers from existing OpenAPI specifications.
icon: code-branch
---
If you have existing REST APIs documented with the OpenAPI Specification (OAS), FastMCP can automatically generate MCP tools, resources, and resource templates directly from that specification. This provides a quick way to make your existing HTTP APIs accessible to MCP clients and LLMs.
## The Goal: API -> MCP Server
The core idea is to map OpenAPI paths and operations (like `GET /users/{id}` or `POST /orders`) to their corresponding MCP components:
- `GET` requests often map to MCP **Resources** (for fetching single items) or **Resource Templates** (if the path has parameters).
- `POST`, `PUT`, `PATCH`, `DELETE` requests typically map to MCP **Tools** (for actions that create or modify data).
FastMCP automates this mapping process.
## Creating from OpenAPI Spec
Use the `FastMCP.from_openapi()` class method. You need:
1. The OpenAPI specification as a Python dictionary.
2. An `httpx.AsyncClient` configured to make requests to the actual API backend.
<CodeGroup>
```python server.py
import asyncio
import httpx
from fastmcp import FastMCP
# load the OpenAPI specification from the openapi_spec.py file
petstore_spec = PETSTORE_SPEC
# Client to communicate with the actual Pet Store API backend
# The base_url should match the server URL in the OpenAPI spec
http_client = httpx.AsyncClient(base_url="http://petstore.example.com/api")
# Create the FastMCP server from the spec
# This is an async class method
async def create_openapi_server():
mcp_server = await FastMCP.from_openapi(
openapi_spec=petstore_spec,
client=http_client,
name="PetStoreMCP" # Optional name for the MCP server
)
return mcp_server
async def run_server():
server = await create_openapi_server()
print(f"Starting OpenAPI-based server '{server.name}'...")
# List discovered components
tools = await server.list_tools()
resources = await server.list_resources()
templates = await server.list_resource_templates()
print("Discovered Tools:", [t.name for t in tools])
print("Discovered Resources:", [r.uri for r in resources]) # Should be empty if no parameterless GETs
print("Discovered Templates:", [t.uriTemplate for t in templates])
# Run the server (e.g., via stdio)
# server.run()
if __name__ == "__main__":
# Example: Create the server and print discovered components
# Requires httpx: uv pip install httpx
asyncio.run(run_server())
# Expected Output might include:
# Discovered Tools: ['listPets', 'createPet']
# Discovered Resources: []
# Discovered Templates: ['resource://openapi/showPetById/{petId}']
```
```python openapi_spec.py
# Example OpenAPI Specification (simplified Pet Store)
PETSTORE_SPEC = {
"openapi": "3.1.0",
"info": {"title": "Simple Pet Store", "version": "1.0.0"},
"servers": [{"url": "http://petstore.example.com/api"}], # Base URL for API calls
"paths": {
"/pets": {
"get": {
"summary": "List all pets",
"operationId": "listPets",
"tags": ["pets"],
"parameters": [{ # Query parameter -> Tool argument
"name": "limit", "in": "query", "schema": {"type": "integer"}
}],
"responses": {"200": {"description": "A list of pets."}},
},
"post": { # POST -> Tool
"summary": "Create a pet",
"operationId": "createPet",
"tags": ["pets"],
"requestBody": { # Request body -> Tool arguments
"required": True,
"content": {"application/json": {"schema": {"$ref": "#/components/schemas/PetInput"}}}
},
"responses": {"201": {"description": "Pet created."}},
},
},
"/pets/{petId}": { # Path parameter -> Resource Template
"get": { # GET with path param -> Resource Template / FunctionResource
"summary": "Info for a specific pet",
"operationId": "showPetById",
"tags": ["pets"],
"parameters": [{ # Path parameter -> Template function argument
"name": "petId", "in": "path", "required": True, "schema": {"type": "string"}
}],
"responses": {"200": {"description": "Information about the pet."}},
},
},
},
"components": {
"schemas": {
"PetInput": {"type": "object", "properties": {"name": {"type": "string"}, "tag": {"type": "string"}}},
}
}
}
```
</CodeGroup>
### How it Works Internally
1. **Parsing**: `from_openapi` parses the spec using utilities that leverage `openapi-pydantic`. It extracts paths, operations, parameters, request bodies, and responses.
2. **Mapping**: It applies mapping rules (see below) to decide whether each OpenAPI operation (`GET /pets`, `POST /pets`, `GET /pets/{petId}`) becomes an MCP `Tool`, `Resource`, or `ResourceTemplate`.
3. **Component Creation**: It creates specialized internal components (`OpenAPITool`, `OpenAPIResource`, `OpenAPIResourceTemplate`).
4. **HTTP Execution**: When an MCP client calls a tool or reads a resource from this server:
* The corresponding OpenAPI component constructs an HTTP request based on the OpenAPI definition and the arguments provided by the MCP client.
* It uses the provided `httpx.AsyncClient` to send the request to the backend API.
* It processes the HTTP response and returns it to the MCP client in the appropriate MCP format.
5. **Schema Generation**: The schemas for MCP tools are derived by combining OpenAPI parameters (path, query, header) and request body schemas. Resource template function arguments are derived from path parameters.
6. **Descriptions**: Tool/Resource descriptions are enhanced with information from OpenAPI responses to give the LLM more context about potential outcomes.
### Default Mapping Rules
FastMCP uses the following default rules to map OpenAPI operations:
- `GET` operation with path parameters (e.g., `/users/{id}`) -> **`ResourceTemplate`**
- `GET` operation without path parameters (e.g., `/users`) -> **`Resource`**
- `POST`, `PUT`, `PATCH`, `DELETE`, `OPTIONS`, `HEAD` -> **`Tool`**
### Customize Route Mapping
You can customize the mapping rules by providing a list of `RouteMap` objects directly to `FastMCP.from_openapi()` using the `route_maps` parameter:
```python
from fastmcp.server.openapi import RouteMap, RouteType
from fastmcp import FastMCP
# Custom mapping: Treat GET /admin/stats as a Tool, not a Resource
custom_maps = [
RouteMap(methods=["GET"], pattern=r"^/admin/stats$", route_type=RouteType.TOOL)
]
async def create_server_with_custom_mapping():
mcp_server = await FastMCP.from_openapi(
openapi_spec=petstore_spec,
client=http_client,
name="PetStoreMCP",
route_maps=custom_maps # Pass custom mapping rules
)
return mcp_server
```
Each `RouteMap` maps one or more HTTP methods and a regular expression pattern for the route path to an MCP `RouteType`. Route maps are processed in order, and the first match wins.
All parameters passed to `FastMCP.from_openapi()` will be forwarded to the underlying `FastMCPOpenAPI` constructor, so you can customize any aspect of the OpenAPI integration directly through this method call.

109
docs/patterns/proxying.mdx Normal file
View file

@ -0,0 +1,109 @@
---
title: Proxying Servers
sidebarTitle: Proxying
description: Use FastMCP to act as an intermediary or change transport for other MCP servers.
icon: arrows-retweet
---
FastMCP provides a powerful proxying capability that allows one FastMCP server instance to act as a frontend for another MCP server (which could be remote, running on a different transport, or even another FastMCP instance). This is achieved using the `FastMCP.as_proxy()` class method.
## What is Proxying?
Proxying means setting up a FastMCP server that doesn't implement its own tools or resources directly. Instead, when it receives a request (like `tools/call` or `resources/read`), it forwards that request to a *backend* MCP server, receives the response, and then relays that response back to the original client.
```mermaid
sequenceDiagram
participant Client
participant ProxyServer as FastMCP Proxy Server
participant BackendServer as Backend MCP Server
Client->>ProxyServer: Request (e.g., stdio)
ProxyServer->>BackendServer: Request (e.g., sse)
BackendServer-->>ProxyServer: Response (e.g., sse)
ProxyServer-->>Client: Response (e.g., stdio)
```
### Use Cases
- **Transport Bridging**: Expose a server running on one transport (e.g., a remote SSE server) via a different transport (e.g., local Stdio for Claude Desktop).
- **Adding Functionality**: Insert a layer in front of an existing server to add caching, logging, authentication, or modify requests/responses (though direct modification requires subclassing `FastMCPProxy`).
- **Security Boundary**: Use the proxy as a controlled gateway to an internal server.
- **Simplifying Client Configuration**: Provide a single, stable endpoint (the proxy) even if the backend server's location or transport changes.
## Creating a Proxy
The easiest way to create a proxy is using the `FastMCP.as_proxy()` class method. This creates a standard FastMCP server that forwards requests to another MCP server.
```python
from fastmcp import FastMCP, Client
# Create a client configured to talk to the backend server
# This could be any MCP server - remote, local, or using any transport
backend_client = Client("backend_server.py") # Could be "http://remote.server/sse", etc.
# Create the proxy server with as_proxy()
proxy_server = await FastMCP.as_proxy(
backend_client,
name="MyProxyServer" # Optional settings for the proxy
)
# That's it! You now have a proxy FastMCP server that can be used
# with any transport (SSE, stdio, etc.) just like any other FastMCP server
```
**How `as_proxy` Works:**
1. It connects to the backend server using the provided client.
2. It discovers all the tools, resources, resource templates, and prompts available on the backend server.
3. It creates corresponding "proxy" components that forward requests to the backend.
4. It returns a standard `FastMCP` server instance that can be used like any other.
### Bridging Transports
A common use case is to bridge transports. For example, making a remote SSE server available locally via Stdio:
```python
from fastmcp import FastMCP, Client
# Client targeting a remote SSE server
client = Client("http://example.com/mcp/sse")
# Create a proxy server - it's just a regular FastMCP server
proxy = await FastMCP.as_proxy(client, name="SSE to Stdio Proxy")
# The proxy can now be used with any transport
# No special handling needed - it works like any FastMCP server
```
### In-Memory Proxies
You can also proxy an in-memory `FastMCP` instance, which is useful for adjusting the configuration or behavior of a server you don't completely control.
```python
from fastmcp import FastMCP
# Original server
original_server = FastMCP(name="Original")
@original_server.tool()
def tool_a() -> str:
return "A"
# Create a proxy of the original server
proxy = await FastMCP.as_proxy(
original_server,
name="Proxy Server"
)
# proxy is now a regular FastMCP server that forwards
# requests to original_server
```
## `FastMCPProxy` Class
Internally, `FastMCP.as_proxy()` uses the `FastMCPProxy` class. You generally don't need to interact with this class directly, but it's available if needed. It has two primary async constructors:
* `FastMCPProxy.from_client(client: Client, **settings)`: Creates a proxy from a client instance.
* `FastMCPProxy.from_server(server: FastMCP, **settings)`: Creates a proxy from another FastMCP server instance.
Using the class directly might be necessary for advanced scenarios, like subclassing `FastMCPProxy` to add custom logic before or after forwarding requests.