mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
Add support for RouteMap tags, update docs
This commit is contained in:
parent
702412e28b
commit
6cac09cf2a
9 changed files with 585 additions and 931 deletions
|
|
@ -50,6 +50,7 @@
|
|||
"servers/resources",
|
||||
"servers/prompts",
|
||||
"servers/context",
|
||||
"servers/openapi",
|
||||
"servers/proxy",
|
||||
"servers/composition"
|
||||
]
|
||||
|
|
@ -76,8 +77,6 @@
|
|||
"pages": [
|
||||
"patterns/decorating-methods",
|
||||
"patterns/http-requests",
|
||||
"patterns/openapi",
|
||||
"patterns/fastapi",
|
||||
"patterns/contrib",
|
||||
"patterns/testing"
|
||||
]
|
||||
|
|
|
|||
|
|
@ -8,19 +8,18 @@ import { VersionBadge } from '/snippets/version-badge.mdx'
|
|||
|
||||
<VersionBadge version="2.0.0" />
|
||||
|
||||
<Note>
|
||||
**Documentation Moved**: The comprehensive FastAPI integration documentation has been moved to the [OpenAPI Integration](/patterns/openapi#fastapi-integration) page, where it's covered alongside all other OpenAPI features including route mapping and tags support.
|
||||
</Note>
|
||||
|
||||
FastMCP can automatically convert FastAPI applications into MCP servers.
|
||||
## Quick Start
|
||||
|
||||
<Tip>
|
||||
FastMCP does *not* include FastAPI as a dependency; you must install it separately to run these examples.
|
||||
</Tip>
|
||||
FastMCP can automatically convert FastAPI applications into MCP servers:
|
||||
|
||||
|
||||
```python {2, 22, 25}
|
||||
```python
|
||||
from fastapi import FastAPI
|
||||
from fastmcp import FastMCP
|
||||
|
||||
|
||||
# A FastAPI app
|
||||
app = FastAPI()
|
||||
|
||||
|
|
@ -36,7 +35,6 @@ def get_item(item_id: int):
|
|||
def create_item(name: str):
|
||||
return {"id": 3, "name": name}
|
||||
|
||||
|
||||
# Create an MCP server from your FastAPI app
|
||||
mcp = FastMCP.from_fastapi(app=app)
|
||||
|
||||
|
|
@ -44,101 +42,6 @@ if __name__ == "__main__":
|
|||
mcp.run() # Start the MCP server
|
||||
```
|
||||
|
||||
## Configuration Options
|
||||
|
||||
### Timeout
|
||||
|
||||
You can set a timeout for all API requests:
|
||||
|
||||
```python
|
||||
# Set a 5 second timeout for all requests
|
||||
mcp = FastMCP.from_fastapi(app=app, timeout=5.0)
|
||||
```
|
||||
|
||||
This timeout is applied to all requests made by tools, resources, and resource templates.
|
||||
|
||||
## Route Mapping
|
||||
|
||||
By default, FastMCP will map FastAPI routes to MCP components according to the following rules:
|
||||
|
||||
| FastAPI Route Type | FastAPI Example | MCP Component | Notes |
|
||||
|--------------------|--------------|---------|-------|
|
||||
| GET without path params | `@app.get("/stats")` | Resource | Simple resources for fetching data |
|
||||
| GET with path params | `@app.get("/users/{id}")` | Resource Template | Path parameters become template parameters |
|
||||
| POST, PUT, DELETE, etc. | `@app.post("/users")` | Tool | Operations that modify data |
|
||||
|
||||
For more details on route mapping or custom mapping rules, see the [OpenAPI integration documentation](/patterns/openapi#route-mapping); FastMCP uses the same mapping rules for both FastAPI and OpenAPI integrations.
|
||||
|
||||
## Complete Example
|
||||
|
||||
Here's a more detailed example with a data model:
|
||||
|
||||
```python [expandable]
|
||||
import asyncio
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from pydantic import BaseModel
|
||||
from fastmcp import FastMCP, Client
|
||||
|
||||
# Define your Pydantic model
|
||||
class Item(BaseModel):
|
||||
name: str
|
||||
price: float
|
||||
|
||||
# Create your FastAPI app
|
||||
app = FastAPI()
|
||||
items = {} # In-memory database
|
||||
|
||||
@app.get("/items")
|
||||
def list_items():
|
||||
"""List all items"""
|
||||
return list(items.values())
|
||||
|
||||
@app.get("/items/{item_id}")
|
||||
def get_item(item_id: int):
|
||||
"""Get item by ID"""
|
||||
if item_id not in items:
|
||||
raise HTTPException(404, "Item not found")
|
||||
return items[item_id]
|
||||
|
||||
@app.post("/items")
|
||||
def create_item(item: Item):
|
||||
"""Create a new item"""
|
||||
item_id = len(items) + 1
|
||||
items[item_id] = {"id": item_id, **item.model_dump()}
|
||||
return items[item_id]
|
||||
|
||||
# Test your MCP server with a client
|
||||
async def check_mcp(mcp: FastMCP):
|
||||
# List the components that were created
|
||||
tools = await mcp.get_tools()
|
||||
resources = await mcp.get_resources()
|
||||
templates = await mcp.get_resource_templates()
|
||||
|
||||
print(
|
||||
f"{len(tools)} Tool(s): {', '.join([t.name for t in tools.values()])}"
|
||||
)
|
||||
print(
|
||||
f"{len(resources)} Resource(s): {', '.join([r.name for r in resources.values()])}"
|
||||
)
|
||||
print(
|
||||
f"{len(templates)} Resource Template(s): {', '.join([t.name for t in templates.values()])}"
|
||||
)
|
||||
|
||||
return mcp
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Create MCP server from FastAPI app
|
||||
mcp = FastMCP.from_fastapi(app=app)
|
||||
|
||||
asyncio.run(check_mcp(mcp))
|
||||
|
||||
# In a real scenario, you would run the server:
|
||||
mcp.run()
|
||||
```
|
||||
|
||||
## Benefits
|
||||
|
||||
- **Leverage existing FastAPI apps** - No need to rewrite your API logic
|
||||
- **Schema reuse** - FastAPI's Pydantic models and validation are inherited
|
||||
- **Full feature support** - Works with FastAPI's authentication, dependencies, etc.
|
||||
- **ASGI transport** - Direct communication without additional HTTP overhead
|
||||
<Tip>
|
||||
For complete documentation including tag-based routing, route mapping configuration, timeout settings, authentication examples, and advanced configuration options, see the comprehensive [OpenAPI Integration documentation](/patterns/openapi#fastapi-integration).
|
||||
</Tip>
|
||||
|
|
@ -1,312 +0,0 @@
|
|||
---
|
||||
title: OpenAPI Integration
|
||||
sidebarTitle: OpenAPI
|
||||
description: Generate MCP servers from OpenAPI specs
|
||||
icon: code-branch
|
||||
---
|
||||
import { VersionBadge } from '/snippets/version-badge.mdx'
|
||||
|
||||
<VersionBadge version="2.0.0" />
|
||||
|
||||
FastMCP can automatically generate an MCP server from an OpenAPI specification. Users only need to provide an OpenAPI specification (3.0 or 3.1) and an API client.
|
||||
|
||||
```python
|
||||
import httpx
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create a client for your API
|
||||
api_client = httpx.AsyncClient(base_url="https://api.example.com")
|
||||
|
||||
# Load your OpenAPI spec
|
||||
spec = {...}
|
||||
|
||||
# Create an MCP server from your OpenAPI spec
|
||||
mcp = FastMCP.from_openapi(openapi_spec=spec, client=api_client)
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run()
|
||||
```
|
||||
|
||||
## Configuration Options
|
||||
|
||||
### Timeout
|
||||
|
||||
You can set a timeout for all requests by providing a `timeout` parameter (in seconds):
|
||||
|
||||
```python
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
timeout=30.0 # 30 second timeout
|
||||
)
|
||||
```
|
||||
|
||||
## Route Mapping
|
||||
|
||||
<VersionBadge version="2.5.0" />
|
||||
|
||||
By default, OpenAPI routes are mapped to MCP components based on these rules:
|
||||
|
||||
| OpenAPI Route | Example |MCP Component | Notes |
|
||||
|- | - | - | - |
|
||||
| `GET` without path params | `GET /stats` | Resource | Simple resources for fetching data |
|
||||
| `GET` with path params | `GET /users/{id}` | Resource Template | Path parameters become template parameters |
|
||||
| `POST`, `PUT`, `PATCH`, `DELETE`, etc. | `POST /users` | Tool | Operations that modify data |
|
||||
|
||||
Internally, FastMCP uses a priority-ordered set of `RouteMap` objects to determine the component type. Route maps indicate that a specific HTTP method (or methods) and path pattern should be treated as a specific component type. This is the default set of route maps:
|
||||
|
||||
```python
|
||||
# Simplified version of the actual mapping rules
|
||||
DEFAULT_ROUTE_MAPPINGS = [
|
||||
# GET with path parameters -> ResourceTemplate
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r".*\{.*\}.*",
|
||||
mcp_type=MCPType.RESOURCE_TEMPLATE,
|
||||
),
|
||||
|
||||
# GET without path parameters -> Resource
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r".*",
|
||||
mcp_type=MCPType.RESOURCE,
|
||||
),
|
||||
|
||||
# All other methods -> Tool
|
||||
ALL_TOOLS(),
|
||||
]
|
||||
```
|
||||
|
||||
#### Custom Route Maps
|
||||
|
||||
Users can add custom route maps to override the default mapping behavior. User-supplied route maps are always applied first, before the default route maps.
|
||||
|
||||
```python
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
# Custom mapping rules
|
||||
custom_maps = [
|
||||
# Force all analytics endpoints to be Tools
|
||||
RouteMap(methods=["GET"],
|
||||
pattern=r"^/analytics/.*",
|
||||
mcp_type=MCPType.TOOL)
|
||||
]
|
||||
|
||||
# Apply custom mappings
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
route_maps=custom_maps
|
||||
)
|
||||
```
|
||||
|
||||
<Info>
|
||||
For backward compatibility, FastMCP still supports the `route_type` parameter and `RouteType` enum, but they are deprecated and will be removed in a future version. You will see deprecation warnings if you use them.
|
||||
</Info>
|
||||
|
||||
#### All Routes as Tools
|
||||
|
||||
When building AI agent backends, it's often useful to treat all routes as callable tools regardless of their HTTP method. You can use the `ALL_TOOLS()` shortcut or create a custom route map:
|
||||
|
||||
```python
|
||||
# Make all endpoints tools using the shortcut
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
route_maps=[ALL_TOOLS()]
|
||||
)
|
||||
|
||||
# Same effect using a custom route map
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
route_maps=[
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL)
|
||||
]
|
||||
)
|
||||
```
|
||||
|
||||
#### Excluding Routes
|
||||
|
||||
If you want to exclude certain routes from being converted to MCP components, you can map them to `MCPType.EXCLUDE`. This is useful for endpoints that should not be accessible to the agent.
|
||||
|
||||
```python
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
# Custom mapping rules to exclude specific routes
|
||||
custom_maps = [
|
||||
# Exclude all admin endpoints
|
||||
RouteMap(
|
||||
methods="*",
|
||||
pattern=r"^/admin/.*",
|
||||
mcp_type=MCPType.EXCLUDE
|
||||
),
|
||||
# Exclude analytics GET endpoints
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r"^/analytics/.*",
|
||||
mcp_type=MCPType.EXCLUDE
|
||||
)
|
||||
]
|
||||
|
||||
# Apply custom mappings
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
route_maps=custom_maps
|
||||
)
|
||||
```
|
||||
|
||||
When a route is mapped to `MCPType.EXCLUDE`, FastMCP will log its presence but won't create any MCP component for it, effectively making it invisible to clients and agents using the MCP server.
|
||||
|
||||
You can customize this behavior by providing a list of `RouteMap` objects:
|
||||
|
||||
```python
|
||||
from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, MCPType
|
||||
|
||||
# Custom route mappings
|
||||
custom_mappings = [
|
||||
# Convert all user-related routes to tools
|
||||
RouteMap(
|
||||
methods=["GET", "POST", "PUT", "DELETE"],
|
||||
pattern=r"^/users.*",
|
||||
mcp_type=MCPType.TOOL
|
||||
),
|
||||
# Exclude analytics routes
|
||||
RouteMap(
|
||||
methods=["*"], # All methods
|
||||
pattern=r"^/analytics.*",
|
||||
mcp_type=MCPType.EXCLUDE
|
||||
),
|
||||
]
|
||||
|
||||
# Create server with custom mappings
|
||||
mcp = FastMCPOpenAPI(
|
||||
openapi_spec=spec,
|
||||
client=httpx.AsyncClient(),
|
||||
route_maps=custom_mappings,
|
||||
)
|
||||
```
|
||||
|
||||
#### Route Map Shortcuts
|
||||
|
||||
|
||||
FastMCP provides several shortcut functions to create common route maps more easily:
|
||||
|
||||
```python
|
||||
from fastmcp.server.openapi import (
|
||||
ALL_TOOLS,
|
||||
EXCLUDE_ALL,
|
||||
EXCLUDE_PATTERN,
|
||||
PATTERN_AS_TOOLS,
|
||||
)
|
||||
|
||||
# Create an MCP server with custom route maps using shortcuts
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
route_maps=[
|
||||
# First exclude all admin endpoints
|
||||
EXCLUDE_PATTERN(r"^/admin/.*"),
|
||||
|
||||
# Make all /api/v1 endpoints tools
|
||||
PATTERN_AS_TOOLS(r"^/api/v1/.*"),
|
||||
|
||||
# Make all remaining routes tools
|
||||
ALL_TOOLS(),
|
||||
]
|
||||
)
|
||||
```
|
||||
|
||||
Available shortcuts:
|
||||
|
||||
| Shortcut Function | Description |
|
||||
|------------------|-------------|
|
||||
| `ALL_TOOLS()` | Converts all matching routes to tools |
|
||||
| `EXCLUDE_ALL()` | Excludes all matching routes from being converted to any component |
|
||||
| `PATTERN_AS_TOOLS(pattern)` | Converts routes matching a specific pattern to tools |
|
||||
| `EXCLUDE_PATTERN(pattern)` | Excludes routes matching a specific pattern |
|
||||
|
||||
These shortcuts are particularly useful for:
|
||||
|
||||
1. Converting all remaining unmatched routes to tools (use `ALL_TOOLS()`)
|
||||
2. Excluding whole sections of your API (use `EXCLUDE_PATTERN("/path/.*")`)
|
||||
3. Converting routes matching specific patterns to tools (use `PATTERN_AS_TOOLS("/path/.*")`)
|
||||
|
||||
<Tip>
|
||||
You can use `EXCLUDE_ALL()` as the last entry in your custom route maps to completely ignore the default route maps. Since custom route maps are applied first and default maps are appended afterward, having `EXCLUDE_ALL()` at the end of your custom maps will match any routes that your earlier custom rules didn't match, preventing the default maps from having any effect.
|
||||
|
||||
```python
|
||||
# Create server that only uses custom route maps, ignoring defaults
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
route_maps=[
|
||||
# Routes to keep as tools
|
||||
PATTERN_AS_TOOLS(r"^/api/v1/.*"),
|
||||
|
||||
# Exclude everything else (ignores default route maps)
|
||||
EXCLUDE_ALL(),
|
||||
]
|
||||
)
|
||||
```
|
||||
</Tip>
|
||||
|
||||
## How It Works
|
||||
|
||||
1. FastMCP parses your OpenAPI spec to extract routes and schemas
|
||||
2. It applies mapping rules to categorize each route
|
||||
3. When an MCP client calls a tool or accesses a resource:
|
||||
- FastMCP constructs an HTTP request based on the OpenAPI definition
|
||||
- It sends the request through the provided httpx client
|
||||
- It translates the HTTP response to the appropriate MCP format
|
||||
|
||||
### Request Parameter Handling
|
||||
|
||||
FastMCP carefully handles different types of parameters in OpenAPI requests:
|
||||
|
||||
#### Query Parameters
|
||||
|
||||
By default, FastMCP will only include query parameters that have non-empty values. Parameters with `None` values or empty strings (`""`) are automatically filtered out of requests. This ensures that API servers don't receive unnecessary empty parameters that might cause issues.
|
||||
|
||||
For example, if you call a tool with these parameters:
|
||||
```python
|
||||
await client.call_tool("search_products", {
|
||||
"category": "electronics", # Will be included
|
||||
"min_price": 100, # Will be included
|
||||
"max_price": None, # Will be excluded
|
||||
"brand": "", # Will be excluded
|
||||
})
|
||||
```
|
||||
|
||||
The resulting HTTP request will only include `category=electronics&min_price=100`.
|
||||
|
||||
#### Path Parameters
|
||||
|
||||
For path parameters, which are typically required by REST APIs, FastMCP filters out `None` values and checks that all required path parameters are provided. If a required path parameter is missing or `None`, an error will be raised.
|
||||
|
||||
```python
|
||||
# This will work
|
||||
await client.call_tool("get_product", {"product_id": 123})
|
||||
|
||||
# This will raise ValueError: "Missing required path parameters: {'product_id'}"
|
||||
await client.call_tool("get_product", {"product_id": None})
|
||||
```
|
||||
|
||||
## Example: Custom Authentication
|
||||
|
||||
If your API requires authentication, you can set headers on the client:
|
||||
|
||||
```python
|
||||
import httpx
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create a client with authentication
|
||||
api_client = httpx.AsyncClient(
|
||||
base_url="https://api.example.com",
|
||||
headers={"Authorization": "Bearer YOUR_TOKEN"}
|
||||
)
|
||||
|
||||
# Create an MCP server from your OpenAPI spec
|
||||
mcp = FastMCP.from_openapi(openapi_spec=spec, client=api_client)
|
||||
```
|
||||
250
docs/servers/openapi.mdx
Normal file
250
docs/servers/openapi.mdx
Normal file
|
|
@ -0,0 +1,250 @@
|
|||
---
|
||||
title: OpenAPI Integration
|
||||
sidebarTitle: OpenAPI Integration
|
||||
description: Generate MCP servers from OpenAPI specs
|
||||
icon: code-branch
|
||||
---
|
||||
import { VersionBadge } from '/snippets/version-badge.mdx'
|
||||
|
||||
<VersionBadge version="2.0.0" />
|
||||
|
||||
FastMCP can automatically generate an MCP server from an OpenAPI specification or FastAPI app. Users only need to provide an OpenAPI specification (3.0 or 3.1) and an API client, or their FastAPI app.
|
||||
|
||||
```python
|
||||
import httpx
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create a client for your API
|
||||
api_client = httpx.AsyncClient(base_url="https://api.example.com")
|
||||
|
||||
# Load your OpenAPI spec
|
||||
spec = {...}
|
||||
|
||||
# Create an MCP server from your OpenAPI spec
|
||||
mcp = FastMCP.from_openapi(openapi_spec=spec, client=api_client)
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run()
|
||||
```
|
||||
|
||||
## Route Mapping
|
||||
|
||||
<VersionBadge version="2.5.0" />
|
||||
|
||||
By default, OpenAPI routes are mapped to MCP components based on these rules:
|
||||
|
||||
| OpenAPI Route | Example |MCP Component |
|
||||
| - | - | - |
|
||||
| `GET` with path params | `GET /users/{id}` | Resource Template |
|
||||
| `GET` without path params | `GET /stats` | Resource |
|
||||
| `POST`, `PUT`, `PATCH`, `DELETE`, etc. | `POST /users` | Tool |
|
||||
|
||||
|
||||
Internally, FastMCP uses a priority-ordered list of `RouteMap` objects to determine the component type for each route. Each `RouteMap` specifies:
|
||||
|
||||
- **Methods**: HTTP methods to match (e.g. `["GET", "POST"]` or `"*"` for all)
|
||||
- **Pattern**: Regex pattern to match the route path (e.g. `r"^/users/.*"` or `r".*"` for all)
|
||||
- **Tags**: A set of OpenAPI tags that must all be present (`{}` means all tags)
|
||||
- **MCP type**: What MCP component type to create (the options are `TOOL`, `RESOURCE`, `RESOURCE_TEMPLATE`, `PROMPT`, or `EXCLUDE` to exclude the route from the MCP server)
|
||||
|
||||
Each OpenAPI route is matched against `RouteMap` objects in order, and the **first match wins** to determine the MCP component type. For example, here are the default route mappings, expressed as `RouteMap` objects in priority order:
|
||||
|
||||
```python
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
# Default route mappings
|
||||
DEFAULT_ROUTE_MAPPINGS = [
|
||||
# GET with path parameters -> ResourceTemplate
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r".*\{.*\}.*",
|
||||
tags={},
|
||||
mcp_type=MCPType.RESOURCE_TEMPLATE
|
||||
),
|
||||
# GET without path parameters -> Resource
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r".*",
|
||||
tags={},
|
||||
mcp_type=MCPType.RESOURCE
|
||||
),
|
||||
# All other methods -> Tool
|
||||
RouteMap(
|
||||
methods="*",
|
||||
pattern=r".*",
|
||||
tags={},
|
||||
mcp_type=MCPType.TOOL
|
||||
),
|
||||
]
|
||||
```
|
||||
|
||||
### Custom Route Maps
|
||||
|
||||
You can override the default behavior by providing custom route maps when creating your MCP server. Custom maps are processed **before** the default maps, so they take priority. Each OpenAPI route will be matched against your custom route maps in order, and the first match will determine the MCP component type (or exclusion!).
|
||||
|
||||
```python {1, 6-18}
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
route_maps=[
|
||||
# All GET analytics endpoints should be tools
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r"^/analytics/.*",
|
||||
mcp_type=MCPType.TOOL,
|
||||
),
|
||||
# Exclude all admin endpoints
|
||||
RouteMap(
|
||||
pattern=r"^/admin/.*",
|
||||
mcp_type=MCPType.EXCLUDE,
|
||||
)
|
||||
]
|
||||
)
|
||||
```
|
||||
|
||||
### Treat All Routes as Tools
|
||||
|
||||
To treat all routes as tools, use `RouteMap(mcp_type=MCPType.TOOL)` as your only route map. It will match all routes and create a tool for each.
|
||||
|
||||
### Prevent Default Mappings
|
||||
|
||||
To prevent the default mappings from being applied, add a catch-all exclusion routemap at the end of your custom route maps: `RouteMap(mcp_type=MCPType.EXCLUDE)`. Since it will match all routes, it will exclude any that weren't match by your previous rules and short-circuit the default mappings.
|
||||
|
||||
### Tag-Based Routing
|
||||
|
||||
<VersionBadge version="2.5.0" />
|
||||
|
||||
To filter routes by OpenAPI tags, use `RouteMap(tags={...})`. The route must have ALL of the specified tags to be matched. If no tags are specified, all routes will be matched.
|
||||
|
||||
|
||||
## Request Parameter Handling
|
||||
|
||||
FastMCP carefully handles different types of parameters in OpenAPI requests:
|
||||
|
||||
### Query Parameters
|
||||
|
||||
By default, FastMCP will only include query parameters that have non-empty values. Parameters with `None` values or empty strings (`""`) are automatically filtered out of requests. This ensures that API servers don't receive unnecessary empty parameters that might cause issues.
|
||||
|
||||
For example, if you call a tool with these parameters:
|
||||
```python
|
||||
await client.call_tool("search_products", {
|
||||
"category": "electronics", # Will be included
|
||||
"min_price": 100, # Will be included
|
||||
"max_price": None, # Will be excluded
|
||||
"brand": "", # Will be excluded
|
||||
})
|
||||
```
|
||||
|
||||
The resulting HTTP request will only include `category=electronics&min_price=100`.
|
||||
|
||||
### Path Parameters
|
||||
|
||||
For path parameters, which are typically required by REST APIs, FastMCP filters out `None` values and checks that all required path parameters are provided. If a required path parameter is missing or `None`, an error will be raised.
|
||||
|
||||
```python
|
||||
# This will work
|
||||
await client.call_tool("get_product", {"product_id": 123})
|
||||
|
||||
# This will raise ValueError: "Missing required path parameters: {'product_id'}"
|
||||
await client.call_tool("get_product", {"product_id": None})
|
||||
```
|
||||
|
||||
## Authorization
|
||||
|
||||
If your API requires authentication, set headers on the client before creating the MCP server.
|
||||
|
||||
```python
|
||||
import httpx
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create a client with authentication
|
||||
api_client = httpx.AsyncClient(
|
||||
base_url="https://api.example.com",
|
||||
headers={"Authorization": "Bearer YOUR_TOKEN"}
|
||||
)
|
||||
|
||||
# Create an MCP server from your OpenAPI spec
|
||||
mcp = FastMCP.from_openapi(openapi_spec=spec, client=api_client)
|
||||
```
|
||||
|
||||
## Timeouts
|
||||
|
||||
You can set a timeout for all requests by providing a `timeout` parameter (in seconds):
|
||||
|
||||
```python
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
timeout=30.0 # 30 second timeout
|
||||
)
|
||||
```
|
||||
|
||||
## FastAPI Integration
|
||||
|
||||
<VersionBadge version="2.0.0" />
|
||||
|
||||
FastMCP can automatically convert FastAPI applications into MCP servers by extracting their OpenAPI specifications. A special client will be created that uses an in-memory ASGI transport to avoid network calls to your FastAPI app. Note that the resulting MCP server is *not* a FastAPI app itself, but can be added to one (see [ASGI integration](/deployment/asgi)).
|
||||
|
||||
<Tip>
|
||||
FastMCP does *not* include FastAPI as a dependency; you must install it separately to use this integration.
|
||||
</Tip>
|
||||
|
||||
```python
|
||||
from fastapi import FastAPI
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# A FastAPI app
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/items", tags=["items"])
|
||||
def list_items():
|
||||
return [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]
|
||||
|
||||
@app.get("/items/{item_id}", tags=["items", "detail"])
|
||||
def get_item(item_id: int):
|
||||
return {"id": item_id, "name": f"Item {item_id}"}
|
||||
|
||||
@app.post("/items", tags=["items", "create"])
|
||||
def create_item(name: str):
|
||||
return {"id": 3, "name": name}
|
||||
|
||||
# Create an MCP server from your FastAPI app
|
||||
mcp = FastMCP.from_fastapi(app=app)
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run() # Start the MCP server
|
||||
```
|
||||
|
||||
### Configuration Options
|
||||
|
||||
**Timeout**: You can set a timeout for all API requests:
|
||||
|
||||
```python
|
||||
# Set a 5 second timeout for all requests
|
||||
mcp = FastMCP.from_fastapi(app=app, timeout=5.0)
|
||||
```
|
||||
|
||||
**Route Mapping**: All the route mapping features (including tags) work with FastAPI apps:
|
||||
|
||||
```python
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
# Use tag-based routing with FastAPI
|
||||
mcp = FastMCP.from_fastapi(
|
||||
app=app,
|
||||
route_maps=[
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL, tags={"admin"}),
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.EXCLUDE, tags={"internal"}),
|
||||
]
|
||||
)
|
||||
```
|
||||
|
||||
### Benefits
|
||||
|
||||
- **Leverage existing FastAPI apps** - No need to rewrite your API logic
|
||||
- **Schema reuse** - FastAPI's Pydantic models and validation are inherited
|
||||
- **Full feature support** - Works with FastAPI's authentication, dependencies, etc.
|
||||
- **ASGI transport** - Direct communication without additional HTTP overhead
|
||||
|
||||
141
examples/tags_example.py
Normal file
141
examples/tags_example.py
Normal file
|
|
@ -0,0 +1,141 @@
|
|||
"""
|
||||
Example demonstrating RouteMap tags functionality.
|
||||
|
||||
This example shows how to use the tags parameter in RouteMap
|
||||
to selectively route OpenAPI endpoints based on their tags.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.openapi import MCPType, RouteMap
|
||||
|
||||
# Create a FastAPI app with tagged endpoints
|
||||
app = FastAPI(title="Tagged API Example")
|
||||
|
||||
|
||||
@app.get("/users", tags=["users", "public"])
|
||||
async def get_users():
|
||||
"""Get all users - public endpoint"""
|
||||
return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]
|
||||
|
||||
|
||||
@app.post("/users", tags=["users", "admin"])
|
||||
async def create_user(name: str):
|
||||
"""Create a user - admin only"""
|
||||
return {"id": 3, "name": name}
|
||||
|
||||
|
||||
@app.get("/admin/stats", tags=["admin", "internal"])
|
||||
async def get_admin_stats():
|
||||
"""Get admin statistics - internal use"""
|
||||
return {"total_users": 100, "active_sessions": 25}
|
||||
|
||||
|
||||
@app.get("/health", tags=["public"])
|
||||
async def health_check():
|
||||
"""Public health check"""
|
||||
return {"status": "healthy"}
|
||||
|
||||
|
||||
@app.get("/metrics")
|
||||
async def get_metrics():
|
||||
"""Metrics endpoint with no tags"""
|
||||
return {"requests": 1000, "errors": 5}
|
||||
|
||||
|
||||
async def main():
|
||||
"""Demonstrate different tag-based routing strategies."""
|
||||
|
||||
print("=== Example 1: Make admin-tagged routes tools ===")
|
||||
|
||||
# Strategy 1: Convert admin-tagged routes to tools
|
||||
mcp1 = FastMCP.from_fastapi(
|
||||
app=app,
|
||||
route_maps=[
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL, tags={"admin"}),
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
],
|
||||
)
|
||||
|
||||
tools = await mcp1.get_tools()
|
||||
resources = await mcp1.get_resources()
|
||||
|
||||
print(f"Tools ({len(tools)}): {', '.join(tools.keys())}")
|
||||
print(f"Resources ({len(resources)}): {', '.join(resources.keys())}")
|
||||
|
||||
print("\n=== Example 2: Exclude internal routes ===")
|
||||
|
||||
# Strategy 2: Exclude internal routes entirely
|
||||
mcp2 = FastMCP.from_fastapi(
|
||||
app=app,
|
||||
route_maps=[
|
||||
RouteMap(
|
||||
methods="*", pattern=r".*", mcp_type=MCPType.EXCLUDE, tags={"internal"}
|
||||
),
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
RouteMap(methods=["POST"], pattern=r".*", mcp_type=MCPType.TOOL),
|
||||
],
|
||||
)
|
||||
|
||||
tools = await mcp2.get_tools()
|
||||
resources = await mcp2.get_resources()
|
||||
|
||||
print(f"Tools ({len(tools)}): {', '.join(tools.keys())}")
|
||||
print(f"Resources ({len(resources)}): {', '.join(resources.keys())}")
|
||||
|
||||
print("\n=== Example 3: Pattern + Tags combination ===")
|
||||
|
||||
# Strategy 3: Routes matching both pattern AND tags
|
||||
mcp3 = FastMCP.from_fastapi(
|
||||
app=app,
|
||||
route_maps=[
|
||||
# Admin routes under /admin path -> tools
|
||||
RouteMap(
|
||||
methods="*",
|
||||
pattern=r".*/admin/.*",
|
||||
mcp_type=MCPType.TOOL,
|
||||
tags={"admin"},
|
||||
),
|
||||
# Public routes -> tools
|
||||
RouteMap(
|
||||
methods="*", pattern=r".*", mcp_type=MCPType.TOOL, tags={"public"}
|
||||
),
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
],
|
||||
)
|
||||
|
||||
tools = await mcp3.get_tools()
|
||||
resources = await mcp3.get_resources()
|
||||
|
||||
print(f"Tools ({len(tools)}): {', '.join(tools.keys())}")
|
||||
print(f"Resources ({len(resources)}): {', '.join(resources.keys())}")
|
||||
|
||||
print("\n=== Example 4: Multiple tag AND condition ===")
|
||||
|
||||
# Strategy 4: Routes must have ALL specified tags
|
||||
mcp4 = FastMCP.from_fastapi(
|
||||
app=app,
|
||||
route_maps=[
|
||||
# Routes with BOTH "users" AND "admin" tags -> tools
|
||||
RouteMap(
|
||||
methods="*",
|
||||
pattern=r".*",
|
||||
mcp_type=MCPType.TOOL,
|
||||
tags={"users", "admin"},
|
||||
),
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
],
|
||||
)
|
||||
|
||||
tools = await mcp4.get_tools()
|
||||
resources = await mcp4.get_resources()
|
||||
|
||||
print(f"Tools ({len(tools)}): {', '.join(tools.keys())}")
|
||||
print(f"Resources ({len(resources)}): {', '.join(resources.keys())}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
|
|
@ -76,6 +76,7 @@ class RouteMap:
|
|||
pattern: Pattern[str] | str = field(default=r".*")
|
||||
mcp_type: MCPType | None = field(default=None)
|
||||
route_type: RouteType | MCPType | None = field(default=None)
|
||||
tags: set[str] = field(default_factory=set)
|
||||
|
||||
def __post_init__(self):
|
||||
"""Validate and process the route map after initialization."""
|
||||
|
|
@ -119,57 +120,6 @@ class RouteMap:
|
|||
self.route_type = self.mcp_type
|
||||
|
||||
|
||||
# Common route map pattern functions
|
||||
def EXCLUDE_ALL() -> RouteMap:
|
||||
"""
|
||||
Create a RouteMap that excludes all routes that haven't been matched by earlier rules.
|
||||
|
||||
This is useful as the last route map to exclude any routes that don't match specific patterns.
|
||||
|
||||
Returns:
|
||||
RouteMap: A route map that excludes all routes
|
||||
"""
|
||||
return RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.EXCLUDE)
|
||||
|
||||
|
||||
def ALL_TOOLS() -> RouteMap:
|
||||
"""
|
||||
Create a RouteMap that converts all routes to tools that haven't been matched by earlier rules.
|
||||
|
||||
This is useful to replace the last item in the default route mappings to make all unmatched routes tools.
|
||||
|
||||
Returns:
|
||||
RouteMap: A route map that converts all routes to tools
|
||||
"""
|
||||
return RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL)
|
||||
|
||||
|
||||
def PATTERN_AS_TOOLS(pattern: str) -> RouteMap:
|
||||
"""
|
||||
Create a RouteMap that converts routes matching a specific pattern to tools.
|
||||
|
||||
Args:
|
||||
pattern: Regex pattern to match routes
|
||||
|
||||
Returns:
|
||||
RouteMap: A route map that converts routes matching the pattern to tools
|
||||
"""
|
||||
return RouteMap(methods="*", pattern=pattern, mcp_type=MCPType.TOOL)
|
||||
|
||||
|
||||
def EXCLUDE_PATTERN(pattern: str) -> RouteMap:
|
||||
"""
|
||||
Create a RouteMap that excludes routes matching a specific pattern.
|
||||
|
||||
Args:
|
||||
pattern: Regex pattern to match routes to exclude
|
||||
|
||||
Returns:
|
||||
RouteMap: A route map that excludes routes matching the pattern
|
||||
"""
|
||||
return RouteMap(methods="*", pattern=pattern, mcp_type=MCPType.EXCLUDE)
|
||||
|
||||
|
||||
# Default route mappings as a list, where order determines priority
|
||||
DEFAULT_ROUTE_MAPPINGS = [
|
||||
# GET requests with path parameters go to ResourceTemplate
|
||||
|
|
@ -179,7 +129,7 @@ DEFAULT_ROUTE_MAPPINGS = [
|
|||
# GET requests without path parameters go to Resource
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
# All other HTTP methods go to Tool
|
||||
ALL_TOOLS(),
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL),
|
||||
]
|
||||
|
||||
|
||||
|
|
@ -208,6 +158,15 @@ def _determine_route_type(
|
|||
pattern_matches = re.search(route_map.pattern, route.path)
|
||||
|
||||
if pattern_matches:
|
||||
# Check if tags match (if specified)
|
||||
# If route_map.tags is empty, tags are not matched
|
||||
# If route_map.tags is non-empty, all tags must be present in route.tags (AND condition)
|
||||
if route_map.tags:
|
||||
route_tags_set = set(route.tags or [])
|
||||
if not route_map.tags.issubset(route_tags_set):
|
||||
# Tags don't match, continue to next mapping
|
||||
continue
|
||||
|
||||
# We know mcp_type is not None here due to post_init validation
|
||||
assert route_map.mcp_type is not None
|
||||
logger.debug(
|
||||
|
|
|
|||
|
|
@ -1147,13 +1147,13 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
"""
|
||||
Create a FastMCP server from an OpenAPI specification.
|
||||
"""
|
||||
from .openapi import ALL_TOOLS, FastMCPOpenAPI
|
||||
from .openapi import FastMCPOpenAPI, MCPType, RouteMap
|
||||
|
||||
# Deprecated since 2.5.0
|
||||
if all_routes_as_tools:
|
||||
warnings.warn(
|
||||
"The 'all_routes_as_tools' parameter is deprecated and will be removed in a future version. "
|
||||
"Use 'route_maps=[ALL_TOOLS()]' instead.",
|
||||
'Use \'route_maps=[RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL)]\' instead.',
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
|
@ -1162,7 +1162,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
raise ValueError("Cannot specify both all_routes_as_tools and route_maps")
|
||||
|
||||
elif all_routes_as_tools:
|
||||
route_maps = [ALL_TOOLS()]
|
||||
route_maps = [RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL)]
|
||||
|
||||
return FastMCPOpenAPI(
|
||||
openapi_spec=openapi_spec,
|
||||
|
|
@ -1184,12 +1184,13 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
Create a FastMCP server from a FastAPI application.
|
||||
"""
|
||||
|
||||
from .openapi import ALL_TOOLS, FastMCPOpenAPI
|
||||
from .openapi import FastMCPOpenAPI, MCPType, RouteMap
|
||||
|
||||
# Deprecated since 2.5.0
|
||||
if all_routes_as_tools:
|
||||
warnings.warn(
|
||||
"The 'all_routes_as_tools' parameter is deprecated and will be removed in a future version. "
|
||||
"Use 'route_maps=[ALL_TOOLS()]' instead.",
|
||||
'Use \'route_maps=[RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL)]\' instead.',
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
|
@ -1198,7 +1199,7 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
raise ValueError("Cannot specify both all_routes_as_tools and route_maps")
|
||||
|
||||
elif all_routes_as_tools:
|
||||
route_maps = [ALL_TOOLS()]
|
||||
route_maps = [RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL)]
|
||||
|
||||
client = httpx.AsyncClient(
|
||||
transport=httpx.ASGITransport(app=app), base_url="http://fastapi"
|
||||
|
|
|
|||
|
|
@ -1926,302 +1926,223 @@ class TestRouteMapWildcard:
|
|||
tools = mcp._tool_manager.list_tools()
|
||||
tool_names = {tool.name for tool in tools}
|
||||
|
||||
# Check that all operations were mapped as tools
|
||||
# Check that all 4 operations became tools
|
||||
expected_tools = {"getUsers", "createUser", "getPosts", "createPost"}
|
||||
assert tool_names == expected_tools
|
||||
|
||||
# No resources or templates should be created
|
||||
resources = mcp._resource_manager.get_resources()
|
||||
templates = mcp._resource_manager.get_templates()
|
||||
assert len(resources) == 0
|
||||
assert len(templates) == 0
|
||||
|
||||
async def test_priority_specific_over_wildcard(
|
||||
self, basic_openapi_spec, mock_basic_client
|
||||
):
|
||||
"""Test that specific method maps take priority over wildcard."""
|
||||
# Create route maps with specific method first, then wildcard
|
||||
route_maps = [
|
||||
# GET operations should be mapped to resources
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
# All other operations should be mapped to tools
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL),
|
||||
]
|
||||
|
||||
mcp = FastMCPOpenAPI(
|
||||
openapi_spec=basic_openapi_spec,
|
||||
client=mock_basic_client,
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
# Check GET operations went to resources
|
||||
resources = mcp._resource_manager.get_resources()
|
||||
resource_names = {r.name for r in resources.values()}
|
||||
assert "getUsers" in resource_names
|
||||
assert "getPosts" in resource_names
|
||||
assert len(resources) == 2
|
||||
|
||||
# Check other operations went to tools
|
||||
tools = mcp._tool_manager.list_tools()
|
||||
tool_names = {tool.name for tool in tools}
|
||||
assert "createUser" in tool_names
|
||||
assert "createPost" in tool_names
|
||||
assert len(tools) == 2
|
||||
|
||||
async def test_priority_wildcard_first(self, basic_openapi_spec, mock_basic_client):
|
||||
"""Test that when wildcard is first, it matches everything."""
|
||||
# Create route maps with wildcard first, then specific methods
|
||||
route_maps = [
|
||||
# Wildcard first matches everything
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL),
|
||||
# This should never be reached
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
]
|
||||
|
||||
mcp = FastMCPOpenAPI(
|
||||
openapi_spec=basic_openapi_spec,
|
||||
client=mock_basic_client,
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
# All operations should be tools
|
||||
tools = mcp._tool_manager.list_tools()
|
||||
assert len(tools) == 4
|
||||
|
||||
# No resources should be created
|
||||
resources = mcp._resource_manager.get_resources()
|
||||
assert len(resources) == 0
|
||||
|
||||
async def test_wildcard_with_specific_paths(
|
||||
self, basic_openapi_spec, mock_basic_client
|
||||
):
|
||||
"""Test wildcard methods combined with specific path patterns."""
|
||||
route_maps = [
|
||||
# All methods on /users path -> Resources
|
||||
RouteMap(methods="*", pattern=r".*/users$", mcp_type=MCPType.RESOURCE),
|
||||
# All methods on /posts path -> Tools
|
||||
RouteMap(methods="*", pattern=r".*/posts$", mcp_type=MCPType.TOOL),
|
||||
]
|
||||
|
||||
mcp = FastMCPOpenAPI(
|
||||
openapi_spec=basic_openapi_spec,
|
||||
client=mock_basic_client,
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
# Check /users operations went to resources
|
||||
resources = mcp._resource_manager.get_resources()
|
||||
resource_names = {r.name for r in resources.values()}
|
||||
assert "getUsers" in resource_names
|
||||
assert "createUser" in resource_names
|
||||
assert len(resources) == 2
|
||||
|
||||
# Check /posts operations went to tools
|
||||
tools = mcp._tool_manager.list_tools()
|
||||
tool_names = {tool.name for tool in tools}
|
||||
assert "getPosts" in tool_names
|
||||
assert "createPost" in tool_names
|
||||
assert len(tools) == 2
|
||||
|
||||
|
||||
class TestAllRoutesAsTools:
|
||||
"""Tests for the all_routes_as_tools parameter in FastMCP class methods."""
|
||||
class TestRouteMapTags:
|
||||
"""Tests for RouteMap tags functionality."""
|
||||
|
||||
@pytest.fixture
|
||||
def simple_api_spec(self) -> dict:
|
||||
"""A simple OpenAPI spec with both GET and POST methods."""
|
||||
def tagged_openapi_spec(self) -> dict:
|
||||
"""Create an OpenAPI spec with various tags for testing."""
|
||||
return {
|
||||
"openapi": "3.1.0",
|
||||
"info": {"title": "Test API", "version": "1.0.0"},
|
||||
"info": {"title": "Tagged API", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/items": {
|
||||
"/users": {
|
||||
"get": {
|
||||
"operationId": "getItems",
|
||||
"operationId": "getUsers",
|
||||
"tags": ["users", "public"],
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
},
|
||||
"post": {
|
||||
"operationId": "createItem",
|
||||
"operationId": "createUser",
|
||||
"tags": ["users", "admin"],
|
||||
"responses": {"201": {"description": "Created"}},
|
||||
},
|
||||
},
|
||||
"/admin/stats": {
|
||||
"get": {
|
||||
"operationId": "getAdminStats",
|
||||
"tags": ["admin", "internal"],
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
"/health": {
|
||||
"get": {
|
||||
"operationId": "getHealth",
|
||||
"tags": ["public"],
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
"/metrics": {
|
||||
"get": {
|
||||
"operationId": "getMetrics",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
async def mock_client(self) -> httpx.AsyncClient:
|
||||
"""Simple mock client for testing."""
|
||||
"""Create a simple mock client."""
|
||||
|
||||
async def _responder(request):
|
||||
return httpx.Response(200, json={"result": "ok"})
|
||||
return httpx.Response(200, json={"status": "ok"})
|
||||
|
||||
transport = httpx.MockTransport(_responder)
|
||||
return httpx.AsyncClient(transport=transport, base_url="http://test")
|
||||
|
||||
async def test_from_openapi_all_routes_as_tools(self, simple_api_spec, mock_client):
|
||||
"""Test FastMCP.from_openapi with all_routes_as_tools=True."""
|
||||
async def test_tags_as_tools(self, tagged_openapi_spec, mock_client):
|
||||
"""Test that routes with specific tags are converted to tools."""
|
||||
# Convert routes with "admin" tag to tools
|
||||
route_maps = [
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL, tags={"admin"}),
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
]
|
||||
|
||||
with pytest.warns(DeprecationWarning, match="all_routes_as_tools.*deprecated"):
|
||||
server = FastMCP.from_openapi(
|
||||
openapi_spec=simple_api_spec,
|
||||
client=mock_client,
|
||||
all_routes_as_tools=True,
|
||||
)
|
||||
|
||||
# Check that all routes are tools
|
||||
tools = await server.get_tools()
|
||||
assert len(tools) >= 2 # Should have at least the two endpoints as tools
|
||||
|
||||
# Should have no resources since all routes are tools
|
||||
resources = await server.get_resources()
|
||||
assert len(resources) == 0
|
||||
|
||||
# Should have no resource templates since all routes are tools
|
||||
templates = await server.get_resource_templates()
|
||||
assert len(templates) == 0
|
||||
|
||||
async def test_from_openapi_all_routes_as_tools_conflicting_args(
|
||||
self, simple_api_spec, mock_client
|
||||
):
|
||||
"""Test FastMCP.from_openapi raises error when both route_maps and all_routes_as_tools are provided."""
|
||||
with pytest.raises(
|
||||
ValueError, match="Cannot specify both all_routes_as_tools and route_maps"
|
||||
):
|
||||
with pytest.warns(
|
||||
DeprecationWarning, match="all_routes_as_tools.*deprecated"
|
||||
):
|
||||
FastMCP.from_openapi(
|
||||
openapi_spec=simple_api_spec,
|
||||
client=mock_client,
|
||||
route_maps=[
|
||||
RouteMap(
|
||||
methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE
|
||||
)
|
||||
],
|
||||
all_routes_as_tools=True,
|
||||
)
|
||||
|
||||
async def test_from_fastapi_all_routes_as_tools(self):
|
||||
"""Test FastMCP.from_fastapi with all_routes_as_tools=True."""
|
||||
|
||||
try:
|
||||
import fastapi
|
||||
except ImportError:
|
||||
pytest.skip("FastAPI not available")
|
||||
|
||||
app = fastapi.FastAPI()
|
||||
|
||||
@app.get("/items")
|
||||
def get_items():
|
||||
return {"items": []}
|
||||
|
||||
@app.post("/items")
|
||||
def create_item():
|
||||
return {"item": "created"}
|
||||
|
||||
with pytest.warns(DeprecationWarning, match="all_routes_as_tools.*deprecated"):
|
||||
server = FastMCP.from_fastapi(app=app, all_routes_as_tools=True)
|
||||
|
||||
# Check that all routes are tools
|
||||
tools = await server.get_tools()
|
||||
assert len(tools) >= 2 # Should have at least the two endpoints as tools
|
||||
|
||||
# Should have no resources since all routes are tools
|
||||
resources = await server.get_resources()
|
||||
assert len(resources) == 0
|
||||
|
||||
# Should have no resource templates since all routes are tools
|
||||
templates = await server.get_resource_templates()
|
||||
assert len(templates) == 0
|
||||
|
||||
async def test_from_fastapi_all_routes_as_tools_conflicting_args(self):
|
||||
"""Test FastMCP.from_fastapi raises error when both route_maps and all_routes_as_tools are provided."""
|
||||
try:
|
||||
import fastapi
|
||||
except ImportError:
|
||||
pytest.skip("FastAPI not available")
|
||||
|
||||
app = fastapi.FastAPI()
|
||||
|
||||
with pytest.raises(
|
||||
ValueError, match="Cannot specify both all_routes_as_tools and route_maps"
|
||||
):
|
||||
with pytest.warns(
|
||||
DeprecationWarning, match="all_routes_as_tools.*deprecated"
|
||||
):
|
||||
FastMCP.from_fastapi(
|
||||
app=app,
|
||||
route_maps=[
|
||||
RouteMap(
|
||||
methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE
|
||||
)
|
||||
],
|
||||
all_routes_as_tools=True,
|
||||
)
|
||||
|
||||
|
||||
class TestRouteTypeExclude:
|
||||
@pytest.fixture
|
||||
def basic_openapi_spec(self) -> dict:
|
||||
return {
|
||||
"openapi": "3.0.0",
|
||||
"info": {"title": "Test API", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/items": {
|
||||
"get": {
|
||||
"operationId": "get_items",
|
||||
"summary": "Get all items",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
"/users": {
|
||||
"get": {
|
||||
"operationId": "get_users",
|
||||
"summary": "Get all users",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
"/analytics": {
|
||||
"get": {
|
||||
"operationId": "get_analytics",
|
||||
"summary": "Get analytics data",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
async def mock_client(self) -> httpx.AsyncClient:
|
||||
async def _responder(request):
|
||||
return httpx.Response(200, json={"success": True})
|
||||
|
||||
return httpx.AsyncClient(transport=httpx.MockTransport(_responder))
|
||||
|
||||
async def test_exclude_routes(self, basic_openapi_spec, mock_client):
|
||||
# Create a server with custom mappings that exclude specific routes
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=basic_openapi_spec,
|
||||
openapi_spec=tagged_openapi_spec,
|
||||
client=mock_client,
|
||||
route_maps=[
|
||||
# Exclude analytics endpoints
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r"^/analytics$",
|
||||
mcp_type=MCPType.EXCLUDE,
|
||||
),
|
||||
# Make everything else a resource
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
],
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
# Check that resources were created for non-excluded routes
|
||||
resources = await server.get_resources()
|
||||
resource_uris = [str(r.uri) for r in resources.values()]
|
||||
# Check that admin-tagged routes are tools
|
||||
tools = server._tool_manager.get_tools()
|
||||
tool_names = {t.name for t in tools.values()}
|
||||
|
||||
# The /analytics endpoint should be excluded
|
||||
assert "resource://openapi/get_items" in resource_uris
|
||||
assert "resource://openapi/get_users" in resource_uris
|
||||
assert "resource://openapi/get_analytics" not in resource_uris
|
||||
resources = server._resource_manager.get_resources()
|
||||
resource_names = {r.name for r in resources.values()}
|
||||
|
||||
# Should only have 2 resources (analytics is excluded)
|
||||
assert len(resources) == 2
|
||||
# Routes with "admin" tag should be tools
|
||||
assert "createUser" in tool_names
|
||||
assert "getAdminStats" in tool_names
|
||||
|
||||
# Routes without "admin" tag should be resources
|
||||
assert "getUsers" in resource_names
|
||||
assert "getHealth" in resource_names
|
||||
assert "getMetrics" in resource_names
|
||||
|
||||
async def test_exclude_tags(self, tagged_openapi_spec, mock_client):
|
||||
"""Test that routes with specific tags are excluded."""
|
||||
# Exclude routes with "internal" tag
|
||||
route_maps = [
|
||||
RouteMap(
|
||||
methods="*", pattern=r".*", mcp_type=MCPType.EXCLUDE, tags={"internal"}
|
||||
),
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
RouteMap(methods=["POST"], pattern=r".*", mcp_type=MCPType.TOOL),
|
||||
]
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=tagged_openapi_spec,
|
||||
client=mock_client,
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
# Check that internal-tagged routes are excluded
|
||||
resources = server._resource_manager.get_resources()
|
||||
resource_names = {r.name for r in resources.values()}
|
||||
|
||||
tools = server._tool_manager.get_tools()
|
||||
tool_names = {t.name for t in tools.values()}
|
||||
|
||||
# Internal-tagged route should be excluded
|
||||
assert "getAdminStats" not in resource_names
|
||||
assert "getAdminStats" not in tool_names
|
||||
|
||||
# Other routes should still be present
|
||||
assert "getUsers" in resource_names
|
||||
assert "getHealth" in resource_names
|
||||
assert "getMetrics" in resource_names
|
||||
assert "createUser" in tool_names
|
||||
|
||||
async def test_multiple_tags_and_condition(self, tagged_openapi_spec, mock_client):
|
||||
"""Test that routes must have ALL specified tags (AND condition)."""
|
||||
# Routes must have BOTH "users" AND "admin" tags
|
||||
route_maps = [
|
||||
RouteMap(
|
||||
methods="*",
|
||||
pattern=r".*",
|
||||
mcp_type=MCPType.TOOL,
|
||||
tags={"users", "admin"},
|
||||
),
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
]
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=tagged_openapi_spec,
|
||||
client=mock_client,
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
tools = server._tool_manager.get_tools()
|
||||
tool_names = {t.name for t in tools.values()}
|
||||
|
||||
resources = server._resource_manager.get_resources()
|
||||
resource_names = {r.name for r in resources.values()}
|
||||
|
||||
# Only createUser has both "users" AND "admin" tags
|
||||
assert "createUser" in tool_names
|
||||
|
||||
# Other routes should be resources
|
||||
assert "getUsers" in resource_names # has "users" but not "admin"
|
||||
assert "getAdminStats" in resource_names # has "admin" but not "users"
|
||||
assert "getHealth" in resource_names
|
||||
assert "getMetrics" in resource_names
|
||||
|
||||
async def test_pattern_and_tags_combination(self, tagged_openapi_spec, mock_client):
|
||||
"""Test that both pattern and tags must be satisfied."""
|
||||
# Routes matching pattern AND having specific tags
|
||||
route_maps = [
|
||||
RouteMap(
|
||||
methods="*",
|
||||
pattern=r".*/admin/.*",
|
||||
mcp_type=MCPType.TOOL,
|
||||
tags={"admin"},
|
||||
),
|
||||
RouteMap(methods=["GET"], pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
RouteMap(methods=["POST"], pattern=r".*", mcp_type=MCPType.TOOL),
|
||||
]
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=tagged_openapi_spec,
|
||||
client=mock_client,
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
tools = server._tool_manager.get_tools()
|
||||
tool_names = {t.name for t in tools.values()}
|
||||
|
||||
resources = server._resource_manager.get_resources()
|
||||
resource_names = {r.name for r in resources.values()}
|
||||
|
||||
# Only getAdminStats matches both /admin/ pattern AND "admin" tag
|
||||
assert "getAdminStats" in tool_names
|
||||
|
||||
# createUser has "admin" tag but doesn't match pattern, so it becomes a tool via POST rule
|
||||
assert "createUser" in tool_names
|
||||
|
||||
# Other routes should be resources (GET)
|
||||
assert "getUsers" in resource_names
|
||||
assert "getHealth" in resource_names
|
||||
assert "getMetrics" in resource_names
|
||||
|
||||
async def test_empty_tags_ignored(self, tagged_openapi_spec, mock_client):
|
||||
"""Test that empty tags set is ignored (matches all routes)."""
|
||||
# Empty tags should match all routes
|
||||
route_maps = [
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL, tags=set()),
|
||||
]
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=tagged_openapi_spec,
|
||||
client=mock_client,
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
tools = server._tool_manager.get_tools()
|
||||
tool_names = {t.name for t in tools.values()}
|
||||
|
||||
# All routes should be tools since empty tags matches everything
|
||||
expected_tools = {
|
||||
"getUsers",
|
||||
"createUser",
|
||||
"getAdminStats",
|
||||
"getHealth",
|
||||
"getMetrics",
|
||||
}
|
||||
assert tool_names == expected_tools
|
||||
|
|
|
|||
|
|
@ -1,208 +0,0 @@
|
|||
"""Tests for the route map shortcut functions."""
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from fastmcp.server.openapi import (
|
||||
ALL_TOOLS,
|
||||
EXCLUDE_ALL,
|
||||
EXCLUDE_PATTERN,
|
||||
PATTERN_AS_TOOLS,
|
||||
FastMCPOpenAPI,
|
||||
MCPType,
|
||||
RouteMap,
|
||||
)
|
||||
|
||||
|
||||
class TestRouteMapShortcuts:
|
||||
"""Tests for the route map shortcut functions."""
|
||||
|
||||
def test_functions_return_correct_route_maps(self):
|
||||
"""Test that each shortcut function returns a RouteMap with the expected properties."""
|
||||
# Test EXCLUDE_ALL
|
||||
exclude_all = EXCLUDE_ALL()
|
||||
assert isinstance(exclude_all, RouteMap)
|
||||
assert exclude_all.methods == "*"
|
||||
assert exclude_all.pattern == ".*"
|
||||
assert exclude_all.mcp_type == MCPType.EXCLUDE
|
||||
|
||||
# Test ALL_TOOLS
|
||||
all_tools = ALL_TOOLS()
|
||||
assert isinstance(all_tools, RouteMap)
|
||||
assert all_tools.methods == "*"
|
||||
assert all_tools.pattern == ".*"
|
||||
assert all_tools.mcp_type == MCPType.TOOL
|
||||
|
||||
# Test PATTERN_AS_TOOLS
|
||||
pattern = r"^/api/.*"
|
||||
pattern_as_tools = PATTERN_AS_TOOLS(pattern)
|
||||
assert isinstance(pattern_as_tools, RouteMap)
|
||||
assert pattern_as_tools.methods == "*"
|
||||
assert pattern_as_tools.pattern == pattern
|
||||
assert pattern_as_tools.mcp_type == MCPType.TOOL
|
||||
|
||||
# Test EXCLUDE_PATTERN
|
||||
pattern = r"^/admin/.*"
|
||||
exclude_pattern = EXCLUDE_PATTERN(pattern)
|
||||
assert isinstance(exclude_pattern, RouteMap)
|
||||
assert exclude_pattern.methods == "*"
|
||||
assert exclude_pattern.pattern == pattern
|
||||
assert exclude_pattern.mcp_type == MCPType.EXCLUDE
|
||||
|
||||
def test_backward_compatibility(self):
|
||||
"""Test that backward compatibility with RouteType and route_type works."""
|
||||
from fastmcp.server.openapi import RouteType
|
||||
|
||||
# Test creating a RouteMap with route_type
|
||||
with pytest.warns(DeprecationWarning):
|
||||
route_map = RouteMap(
|
||||
methods=["GET"], pattern=r".*", route_type=RouteType.TOOL
|
||||
)
|
||||
assert route_map.mcp_type == MCPType.TOOL
|
||||
|
||||
# Test accessing fields on RouteType directly
|
||||
# Note: importing RouteType already causes the deprecation warning,
|
||||
# so we don't need to check for it again here
|
||||
rt = RouteType.RESOURCE
|
||||
assert rt.value == "RESOURCE"
|
||||
assert rt.name == "RESOURCE"
|
||||
|
||||
|
||||
class TestRouteMapShortcutsIntegration:
|
||||
"""Integration tests for the route map shortcut functions with FastMCPOpenAPI."""
|
||||
|
||||
@pytest.fixture
|
||||
def basic_openapi_spec(self) -> dict:
|
||||
"""Create a simple OpenAPI spec for testing."""
|
||||
return {
|
||||
"openapi": "3.0.0",
|
||||
"info": {"title": "Test API", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/items": {
|
||||
"get": {
|
||||
"operationId": "get_items",
|
||||
"summary": "Get all items",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
},
|
||||
"post": {
|
||||
"operationId": "create_item",
|
||||
"summary": "Create an item",
|
||||
"responses": {"201": {"description": "Created"}},
|
||||
},
|
||||
},
|
||||
"/users": {
|
||||
"get": {
|
||||
"operationId": "get_users",
|
||||
"summary": "Get all users",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
},
|
||||
},
|
||||
"/admin": {
|
||||
"get": {
|
||||
"operationId": "get_admin",
|
||||
"summary": "Admin endpoint",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
},
|
||||
},
|
||||
"/items/{item_id}": {
|
||||
"get": {
|
||||
"operationId": "get_item",
|
||||
"summary": "Get an item by ID",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "item_id",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
],
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
async def mock_client(self) -> httpx.AsyncClient:
|
||||
"""Create a mock client for testing."""
|
||||
|
||||
async def _responder(request):
|
||||
return httpx.Response(200, json={"success": True})
|
||||
|
||||
return httpx.AsyncClient(transport=httpx.MockTransport(_responder))
|
||||
|
||||
async def test_all_tools(self, basic_openapi_spec, mock_client):
|
||||
"""Test using ALL_TOOLS() to convert all routes to tools."""
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=basic_openapi_spec,
|
||||
client=mock_client,
|
||||
route_maps=[ALL_TOOLS()],
|
||||
)
|
||||
|
||||
# Check that all routes are tools
|
||||
tools = await server.get_tools()
|
||||
resources = await server.get_resources()
|
||||
templates = await server.get_resource_templates()
|
||||
|
||||
# All 5 routes should be tools
|
||||
assert len(tools) == 5
|
||||
assert len(resources) == 0
|
||||
assert len(templates) == 0
|
||||
|
||||
# Check that all expected tools exist
|
||||
tool_names = [t.name for t in tools.values()]
|
||||
assert "get_items" in tool_names
|
||||
assert "create_item" in tool_names
|
||||
assert "get_users" in tool_names
|
||||
assert "get_admin" in tool_names
|
||||
assert "get_item" in tool_names
|
||||
|
||||
async def test_exclude_pattern(self, basic_openapi_spec, mock_client):
|
||||
"""Test using EXCLUDE_PATTERN() to exclude specific routes."""
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=basic_openapi_spec,
|
||||
client=mock_client,
|
||||
route_maps=[
|
||||
# Exclude admin endpoints
|
||||
EXCLUDE_PATTERN(r"^/admin"),
|
||||
# Make everything else a tool
|
||||
ALL_TOOLS(),
|
||||
],
|
||||
)
|
||||
|
||||
# Check that admin route is excluded
|
||||
tools = await server.get_tools()
|
||||
tool_names = [t.name for t in tools.values()]
|
||||
|
||||
# All routes except admin should be tools
|
||||
assert "get_items" in tool_names
|
||||
assert "create_item" in tool_names
|
||||
assert "get_users" in tool_names
|
||||
assert "get_item" in tool_names
|
||||
assert "get_admin" not in tool_names # This should be excluded
|
||||
|
||||
async def test_pattern_as_tools(self, basic_openapi_spec, mock_client):
|
||||
"""Test using PATTERN_AS_TOOLS() to convert routes matching a pattern to tools."""
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=basic_openapi_spec,
|
||||
client=mock_client,
|
||||
route_maps=[
|
||||
# Make /items routes tools regardless of method
|
||||
PATTERN_AS_TOOLS(r"^/items"),
|
||||
# Make everything else a resource
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.RESOURCE),
|
||||
],
|
||||
)
|
||||
|
||||
# Check that /items routes are tools
|
||||
tools = await server.get_tools()
|
||||
tool_names = [t.name for t in tools.values()]
|
||||
assert "get_items" in tool_names
|
||||
assert "create_item" in tool_names
|
||||
assert "get_item" in tool_names
|
||||
|
||||
# Check that other routes are resources
|
||||
resources = await server.get_resources()
|
||||
resource_names = [r.name for r in resources.values()]
|
||||
assert "get_users" in resource_names
|
||||
assert "get_admin" in resource_names
|
||||
Loading…
Add table
Add a link
Reference in a new issue