mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 05:24:18 +02:00
Merge pull request #566 from jlowin/route_map_fn
Add advanced control of openAPI route creation
This commit is contained in:
Jeremiah Lowin
GitHub
commit
fca5501639
6 changed files with 735 additions and 103 deletions
|
|
@ -15,11 +15,11 @@
|
|||
> [!NOTE]
|
||||
> #### FastMCP 2.0 & The Official MCP SDK
|
||||
>
|
||||
> Recognize the `FastMCP` name? You might have seen the version that was contributed to the [official MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk), which was based on **FastMCP 1.0**.
|
||||
> FastMCP is the standard framework for building MCP servers and clients. FastMCP 1.0 was incorporated into the [official MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk).
|
||||
>
|
||||
> **Welcome to FastMCP 2.0!** This is the actively developed successor, and it significantly expands on 1.0 by introducing powerful client capabilities, server proxying & composition, OpenAPI/FastAPI integration, and more advanced features.
|
||||
> **This is FastMCP 2.0,** the actively maintained version that significantly expands on 1.0's basic server-building capabilities by introducing full client support, server composition, OpenAPI/FastAPI integration, remote server proxying, built-in testing tools, and more.
|
||||
>
|
||||
> FastMCP 2.0 is the recommended path for building modern, powerful MCP applications. Ready to upgrade or get started? Follow the [installation instructions](https://gofastmcp.com/getting-started/installation), which include specific steps for upgrading from the official MCP SDK.
|
||||
> FastMCP 2.0 is the complete toolkit for modern AI applications. Ready to upgrade or get started? Follow the [installation instructions](https://gofastmcp.com/getting-started/installation), which include specific steps for upgrading from the official MCP SDK.
|
||||
|
||||
---
|
||||
|
||||
|
|
|
|||
|
|
@ -24,17 +24,13 @@ if __name__ == "__main__":
|
|||
```
|
||||
|
||||
|
||||
## FastMCP 2.0 and the Official MCP SDK
|
||||
## FastMCP and the Official MCP SDK
|
||||
|
||||
<Tip>
|
||||
Recognize the `FastMCP` name? You might have seen the version that was contributed to the [official MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk), which was based on **FastMCP 1.0**.
|
||||
FastMCP is the standard framework for building MCP servers and clients. FastMCP 1.0 was incorporated into the [official MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk).
|
||||
|
||||
**This is FastMCP 2.0,** the [actively maintained version](https://github.com/jlowin/fastmcp) that significantly expands on 1.0's basic server-building capabilities by introducing full client support, server composition, OpenAPI/FastAPI integration, remote server proxying, built-in testing tools, and more.
|
||||
|
||||
**Welcome to FastMCP 2.0!** This is the [actively developed successor](https://github.com/jlowin/fastmcp), and it significantly expands on 1.0 by introducing powerful client capabilities, server proxying & composition, OpenAPI/FastAPI integration, and more advanced features.
|
||||
|
||||
FastMCP 2.0 is the recommended path for building modern, powerful MCP applications. Ready to upgrade or get started? Follow the [installation instructions](/getting-started/installation), which include specific steps for upgrading.
|
||||
</Tip>
|
||||
|
||||
FastMCP 2.0 is the complete toolkit for modern AI applications. Ready to upgrade or get started? Follow the [installation instructions](/getting-started/installation), which include specific steps for upgrading from the official MCP SDK.
|
||||
|
||||
|
||||
## What is MCP?
|
||||
|
|
|
|||
|
|
@ -1,78 +1,96 @@
|
|||
---
|
||||
title: OpenAPI Integration
|
||||
sidebarTitle: OpenAPI Integration
|
||||
description: Generate MCP servers from OpenAPI specs
|
||||
description: Generate MCP servers from OpenAPI specs and FastAPI apps
|
||||
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.
|
||||
FastMCP can automatically generate an MCP server from an OpenAPI specification or FastAPI app. Instead of manually creating tools and resources, you provide an OpenAPI spec and FastMCP intelligently converts your API endpoints into the appropriate MCP components.
|
||||
|
||||
```python
|
||||
## Quick Start
|
||||
|
||||
To convert an OpenAPI specification to an MCP server, you can use the `FastMCP.from_openapi` class method. This method takes an OpenAPI specification and an async HTTPX client that can be used to make requests to the API, and returns an MCP server.
|
||||
|
||||
Here's an example:
|
||||
```python {11-15}
|
||||
import httpx
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create a client for your API
|
||||
api_client = httpx.AsyncClient(base_url="https://api.example.com")
|
||||
# Create an HTTP client for your API
|
||||
client = httpx.AsyncClient(base_url="https://api.example.com")
|
||||
|
||||
# Load your OpenAPI spec
|
||||
spec = {...}
|
||||
# Load your OpenAPI spec
|
||||
openapi_spec = httpx.get("https://api.example.com/openapi.json").json()
|
||||
|
||||
# Create an MCP server from your OpenAPI spec
|
||||
mcp = FastMCP.from_openapi(openapi_spec=spec, client=api_client)
|
||||
# Create the MCP server
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=openapi_spec,
|
||||
client=client,
|
||||
name="My API Server"
|
||||
)
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run()
|
||||
```
|
||||
|
||||
That's it! Your entire API is now available as an MCP server. Clients can discover and interact with your API endpoints through the MCP protocol, with full schema validation and type safety.
|
||||
|
||||
|
||||
## Route Mapping
|
||||
|
||||
|
||||
|
||||
FastMCP analyzes your API specification and automatically creates MCP components based on HTTP semantics and REST conventions. By default, the following rules are used to determine what MCP component to create for each route:
|
||||
|
||||
| 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** |
|
||||
|
||||
|
||||
|
||||
### Custom Route Maps
|
||||
|
||||
<VersionBadge version="2.5.0" />
|
||||
|
||||
By default, OpenAPI routes are mapped to MCP components based on these rules:
|
||||
FastMCP uses an ordered list of `RouteMap` objects to determine how to map OpenAPI routes to various MCP component types.
|
||||
|
||||
| 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:
|
||||
Each `RouteMap` specifies a combination of methods, patterns, and tags, as well as a corresponding MCP component type. Each OpenAPI route is checked against each `RouteMap` in order, and the first one that matches every criteria is used to determine its converted MCP type. A special type, `EXCLUDE`, can be used to exclude routes from the MCP server entirely.
|
||||
|
||||
- **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. An empty set (`{}`) means no tag filtering, so the route matches regardless of its 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)
|
||||
- **MCP type**: What MCP component type to create (`TOOL`, `RESOURCE`, `RESOURCE_TEMPLATE`, or `EXCLUDE`)
|
||||
|
||||
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:
|
||||
To illustrate this in practice, here are FastMCP's default route mappings as a list of `RouteMap` objects:
|
||||
|
||||
```python
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
# Default route mappings
|
||||
DEFAULT_ROUTE_MAPPINGS = [
|
||||
# GET with path parameters -> ResourceTemplate
|
||||
|
||||
# GET with path parameters → ResourceTemplate
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r".*\{.*\}.*",
|
||||
tags={},
|
||||
mcp_type=MCPType.RESOURCE_TEMPLATE
|
||||
),
|
||||
# GET without path parameters -> Resource
|
||||
|
||||
# GET without path parameters → Resource
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r".*",
|
||||
tags={},
|
||||
mcp_type=MCPType.RESOURCE
|
||||
),
|
||||
# All other methods -> Tool
|
||||
|
||||
# All other methods → Tool
|
||||
RouteMap(
|
||||
methods="*",
|
||||
methods=["*"],
|
||||
pattern=r".*",
|
||||
tags={},
|
||||
mcp_type=MCPType.TOOL
|
||||
),
|
||||
]
|
||||
|
|
@ -80,112 +98,272 @@ DEFAULT_ROUTE_MAPPINGS = [
|
|||
|
||||
### 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!).
|
||||
When creating your FastMCP server, you can customize routing behavior by providing your own list of `RouteMap` objects. Your custom maps are processed before the default route maps, and routes will be assigned to the first matching custom map.
|
||||
|
||||
```python {1, 6-18}
|
||||
For example, the following simple rule will treat every OpenAPI route as a tool:
|
||||
|
||||
```python {7}
|
||||
from fastmcp import FastMCP
|
||||
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(mcp_type=MCPType.TOOL),
|
||||
],
|
||||
)
|
||||
```
|
||||
|
||||
Here is a more complete example that uses custom route maps to convert all `GET` endpoints under `/analytics/` to tools while excluding all admin endpoints and all routes tagged "internal". All other routes will be handled by the default rules:
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
mcp = FastMCP.from_openapi(
|
||||
...,
|
||||
route_maps=[
|
||||
|
||||
# Analytics `GET` endpoints are tools
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r"^/analytics/.*",
|
||||
mcp_type=MCPType.TOOL,
|
||||
),
|
||||
|
||||
# Exclude all admin endpoints
|
||||
RouteMap(
|
||||
pattern=r"^/admin/.*",
|
||||
mcp_type=MCPType.EXCLUDE,
|
||||
)
|
||||
]
|
||||
),
|
||||
|
||||
# Exclude all routes tagged "internal"
|
||||
RouteMap(
|
||||
tags={"internal"},
|
||||
mcp_type=MCPType.EXCLUDE,
|
||||
),
|
||||
],
|
||||
)
|
||||
```
|
||||
|
||||
### Treat All Routes as Tools
|
||||
<Tip>
|
||||
The default route maps are always applied after your custom maps, so you do not have to create route maps for every possible route.
|
||||
</Tip>
|
||||
|
||||
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.
|
||||
### Excluding Routes
|
||||
|
||||
### Prevent Default Mappings
|
||||
To exclude routes from the MCP server, use a route map to assign them to `MCPType.EXCLUDE`.
|
||||
|
||||
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.
|
||||
You can use this to remove sensitive or internal routes by targeting them specifically:
|
||||
|
||||
### Tag-Based Routing
|
||||
```python {7,8}
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
mcp = FastMCP.from_openapi(
|
||||
...,
|
||||
route_maps=[
|
||||
RouteMap(pattern=r"^/admin/.*", mcp_type=MCPType.EXCLUDE),
|
||||
RouteMap(tags={"internal"}, mcp_type=MCPType.EXCLUDE),
|
||||
],
|
||||
)
|
||||
```
|
||||
|
||||
Or you can use a catch-all rule to exclude everything that your maps don't handle explicitly:
|
||||
```python {10}
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
mcp = FastMCP.from_openapi(
|
||||
...,
|
||||
route_maps=[
|
||||
# custom mapping logic goes here
|
||||
...,
|
||||
# exclude all remaining routes
|
||||
RouteMap(mcp_type=MCPType.EXCLUDE),
|
||||
],
|
||||
)
|
||||
```
|
||||
|
||||
<Tip>
|
||||
Using a catch-all exclusion rule will prevent the default route mappings from being applied, since it will match every remaining route. This is useful if you want to explicitly allow-list certain routes.
|
||||
</Tip>
|
||||
|
||||
|
||||
### Advanced Route Mapping
|
||||
|
||||
<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.
|
||||
For advanced use cases that require more complex logic, you can provide a `route_map_fn` callable. After the route map logic is applied, this function is called on each matched route and its assigned MCP component type. It can optionally return a different component type to override the mapped assignment. If it returns `None`, the assigned type is used.
|
||||
|
||||
In addition to more precise targeting of methods, patterns, and tags, this function can access any additional OpenAPI metadata about the route.
|
||||
|
||||
<Tip>
|
||||
The `route_map_fn` **is** called on routes that matched `MCPType.EXCLUDE` in your custom maps, giving you an opportunity to override the exclusion.
|
||||
</Tip>
|
||||
|
||||
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.openapi import RouteMap, MCPType, HTTPRoute
|
||||
|
||||
def custom_route_mapper(route: HTTPRoute, mcp_type: MCPType) -> MCPType | None:
|
||||
"""Advanced route type mapping."""
|
||||
# Convert all admin routes to tools regardless of HTTP method
|
||||
if "/admin/" in route.path:
|
||||
return MCPType.TOOL
|
||||
|
||||
elif "internal" in route.tags:
|
||||
return MCPType.EXCLUDE
|
||||
|
||||
# Convert user detail routes to templates even if they're POST
|
||||
elif route.path.startswith("/users/") and route.method == "POST":
|
||||
return MCPType.RESOURCE_TEMPLATE
|
||||
|
||||
# Use defaults for all other routes
|
||||
return None
|
||||
|
||||
mcp = FastMCP.from_openapi(
|
||||
...,
|
||||
route_map_fn=custom_route_mapper,
|
||||
)
|
||||
```
|
||||
|
||||
## Customizing MCP Components
|
||||
|
||||
<VersionBadge version="2.5.0" />
|
||||
|
||||
By default, FastMCP creates MCP components using a variety of metadata from the OpenAPI spec, such as incorporating the OpenAPI description into the MCP component description.
|
||||
|
||||
At times you may want to modify those MCP components in a variety of ways, such as adding LLM-specific instructions or tags. For fine-grained customization, you can provide a `mcp_component_fn` when creating the MCP server. After each MCP component has been created, this function is called on it and has the opportunity to modify it in-place.
|
||||
|
||||
<Tip>
|
||||
Your `mcp_component_fn` is expected to modify the component in-place, not to return a new component. The result of the function is ignored.
|
||||
</Tip>
|
||||
|
||||
```python {27}
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.openapi import (
|
||||
HTTPRoute,
|
||||
OpenAPITool,
|
||||
OpenAPIResource,
|
||||
OpenAPIResourceTemplate,
|
||||
)
|
||||
|
||||
def customize_components(
|
||||
route: HTTPRoute,
|
||||
component: OpenAPITool | OpenAPIResource | OpenAPIResourceTemplate,
|
||||
) -> None:
|
||||
|
||||
# Add custom tags to all components
|
||||
component.tags.add("openapi")
|
||||
|
||||
# Customize based on component type
|
||||
if isinstance(component, OpenAPITool):
|
||||
component.description = f"🔧 {component.description} (via API)"
|
||||
|
||||
if isinstance(component, OpenAPIResource):
|
||||
component.description = f"📊 {component.description}"
|
||||
component.tags.add("data")
|
||||
|
||||
mcp = FastMCP.from_openapi(
|
||||
...,
|
||||
mcp_component_fn=customize_components,
|
||||
)
|
||||
```
|
||||
|
||||
## Request Parameter Handling
|
||||
|
||||
FastMCP carefully handles different types of parameters in OpenAPI requests:
|
||||
FastMCP intelligently 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.
|
||||
By default, FastMCP only includes query parameters that have non-empty values. Parameters with `None` values or empty strings are automatically filtered out.
|
||||
|
||||
For example, if you call a tool with these parameters:
|
||||
```python
|
||||
# When calling this tool...
|
||||
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
|
||||
"category": "electronics", # ✅ Included
|
||||
"min_price": 100, # ✅ Included
|
||||
"max_price": None, # ❌ Excluded
|
||||
"brand": "", # ❌ Excluded
|
||||
})
|
||||
```
|
||||
|
||||
The resulting HTTP request will only include `category=electronics&min_price=100`.
|
||||
# The HTTP request will be: GET /products?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.
|
||||
Path parameters are typically required by REST APIs. FastMCP:
|
||||
- Filters out `None` values
|
||||
- Validates that all required path parameters are provided
|
||||
- Raises clear errors for missing required parameters
|
||||
|
||||
```python
|
||||
# This will work
|
||||
await client.call_tool("get_product", {"product_id": 123})
|
||||
# ✅ This works
|
||||
await client.call_tool("get_user", {"user_id": 123})
|
||||
|
||||
# This will raise ValueError: "Missing required path parameters: {'product_id'}"
|
||||
await client.call_tool("get_product", {"product_id": None})
|
||||
# ❌ This raises: "Missing required path parameters: {'user_id'}"
|
||||
await client.call_tool("get_user", {"user_id": None})
|
||||
```
|
||||
|
||||
## Authorization
|
||||
### Array Parameters
|
||||
|
||||
If your API requires authentication, set headers on the client before creating the MCP server.
|
||||
FastMCP handles array parameters according to OpenAPI specifications:
|
||||
|
||||
- **Query arrays**: Serialized based on the `explode` parameter (default: `True`)
|
||||
- **Path arrays**: Serialized as comma-separated values (OpenAPI 'simple' style)
|
||||
|
||||
```python
|
||||
# Query array with explode=true (default)
|
||||
# ?tags=red&tags=blue&tags=green
|
||||
|
||||
# Query array with explode=false
|
||||
# ?tags=red,blue,green
|
||||
|
||||
# Path array (always comma-separated)
|
||||
# /items/red,blue,green
|
||||
```
|
||||
|
||||
### Headers
|
||||
|
||||
Header parameters are automatically converted to strings and included in the HTTP request.
|
||||
|
||||
## Auth
|
||||
|
||||
If your API requires authentication, configure it on the HTTP client before creating the MCP server:
|
||||
|
||||
```python
|
||||
import httpx
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# Create a client with authentication
|
||||
# Bearer token 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)
|
||||
# Create MCP server with authenticated client
|
||||
mcp = FastMCP.from_openapi(..., client=api_client)
|
||||
```
|
||||
|
||||
## Timeouts
|
||||
|
||||
You can set a timeout for all requests by providing a `timeout` parameter (in seconds):
|
||||
Set a timeout for all API requests:
|
||||
|
||||
```python
|
||||
mcp = FastMCP.from_openapi(
|
||||
openapi_spec=spec,
|
||||
client=api_client,
|
||||
timeout=30.0 # 30 second timeout
|
||||
timeout=30.0 # 30 second timeout for all requests
|
||||
)
|
||||
```
|
||||
|
||||
|
||||
## 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)).
|
||||
FastMCP can directly convert FastAPI applications into MCP servers by extracting their OpenAPI specifications:
|
||||
|
||||
<Tip>
|
||||
FastMCP does *not* include FastAPI as a dependency; you must install it separately to use this integration.
|
||||
|
|
@ -195,8 +373,8 @@ FastMCP does *not* include FastAPI as a dependency; you must install it separate
|
|||
from fastapi import FastAPI
|
||||
from fastmcp import FastMCP
|
||||
|
||||
# A FastAPI app
|
||||
app = FastAPI()
|
||||
# Your FastAPI app
|
||||
app = FastAPI(title="My API", version="1.0.0")
|
||||
|
||||
@app.get("/items", tags=["items"])
|
||||
def list_items():
|
||||
|
|
@ -210,41 +388,46 @@ def get_item(item_id: int):
|
|||
def create_item(name: str):
|
||||
return {"id": 3, "name": name}
|
||||
|
||||
# Create an MCP server from your FastAPI app
|
||||
# Convert FastAPI app to MCP server
|
||||
mcp = FastMCP.from_fastapi(app=app)
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run() # Start the MCP server
|
||||
mcp.run() # Run as MCP server
|
||||
```
|
||||
|
||||
### Configuration Options
|
||||
<Warning>
|
||||
FastMCP servers are not FastAPI apps, even when created from one. To learn how to deploy them as an ASGI app, see the [ASGI Integration](/deployment/asgi) documentation.
|
||||
</Warning>
|
||||
|
||||
**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:
|
||||
### FastAPI Configuration
|
||||
|
||||
All OpenAPI integration features work with FastAPI apps:
|
||||
|
||||
```python
|
||||
from fastmcp.server.openapi import RouteMap, MCPType
|
||||
|
||||
# Use tag-based routing with FastAPI
|
||||
# Custom route mapping with FastAPI
|
||||
mcp = FastMCP.from_fastapi(
|
||||
app=app,
|
||||
name="My Custom Server",
|
||||
timeout=5.0,
|
||||
route_maps=[
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.TOOL, tags={"admin"}),
|
||||
# Admin endpoints become tools
|
||||
RouteMap(methods="*", pattern=r"^/admin/.*", mcp_type=MCPType.TOOL),
|
||||
# Internal endpoints are excluded
|
||||
RouteMap(methods="*", pattern=r".*", mcp_type=MCPType.EXCLUDE, tags={"internal"}),
|
||||
]
|
||||
],
|
||||
route_map_fn=my_route_mapper,
|
||||
mcp_component_fn=my_component_customizer,
|
||||
)
|
||||
```
|
||||
|
||||
### Benefits
|
||||
### FastAPI 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
|
||||
- **Zero code duplication**: Reuse existing FastAPI endpoints
|
||||
- **Schema inheritance**: Pydantic models and validation are preserved
|
||||
- **ASGI transport**: Direct in-memory communication (no HTTP overhead)
|
||||
- **Full FastAPI features**: Dependencies, middleware, authentication all work
|
||||
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ from fastmcp.tools.tool import Tool, _convert_to_content
|
|||
from fastmcp.utilities import openapi
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.openapi import (
|
||||
HTTPRoute,
|
||||
_combine_schemas,
|
||||
format_description_with_responses,
|
||||
)
|
||||
|
|
@ -33,6 +34,16 @@ logger = get_logger(__name__)
|
|||
|
||||
HttpMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"]
|
||||
|
||||
# Type definitions for the mapping functions
|
||||
RouteMapFn = Callable[[HTTPRoute, "MCPType"], "MCPType | None"]
|
||||
ComponentFn = Callable[
|
||||
[
|
||||
HTTPRoute,
|
||||
"OpenAPITool | OpenAPIResource | OpenAPIResourceTemplate",
|
||||
],
|
||||
None,
|
||||
]
|
||||
|
||||
|
||||
class MCPType(enum.Enum):
|
||||
"""Type of FastMCP component to create from a route.
|
||||
|
|
@ -41,7 +52,6 @@ class MCPType(enum.Enum):
|
|||
TOOL: Convert the route to a callable Tool
|
||||
RESOURCE: Convert the route to a Resource (typically GET endpoints)
|
||||
RESOURCE_TEMPLATE: Convert the route to a ResourceTemplate (typically GET with path params)
|
||||
PROMPT: Convert the route to a Prompt (not yet implemented)
|
||||
EXCLUDE: Exclude the route from being converted to any MCP component
|
||||
IGNORE: Deprecated, use EXCLUDE instead
|
||||
"""
|
||||
|
|
@ -49,7 +59,7 @@ class MCPType(enum.Enum):
|
|||
TOOL = "TOOL"
|
||||
RESOURCE = "RESOURCE"
|
||||
RESOURCE_TEMPLATE = "RESOURCE_TEMPLATE"
|
||||
PROMPT = "PROMPT"
|
||||
# PROMPT = "PROMPT"
|
||||
EXCLUDE = "EXCLUDE"
|
||||
|
||||
|
||||
|
|
@ -64,7 +74,6 @@ class RouteType(enum.Enum):
|
|||
TOOL = "TOOL"
|
||||
RESOURCE = "RESOURCE"
|
||||
RESOURCE_TEMPLATE = "RESOURCE_TEMPLATE"
|
||||
PROMPT = "PROMPT"
|
||||
IGNORE = "IGNORE"
|
||||
|
||||
|
||||
|
|
@ -614,7 +623,7 @@ class FastMCPOpenAPI(FastMCP):
|
|||
|
||||
Example:
|
||||
```python
|
||||
from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, RouteType
|
||||
from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap, MCPType
|
||||
import httpx
|
||||
|
||||
# Define custom route mappings
|
||||
|
|
@ -633,7 +642,7 @@ class FastMCPOpenAPI(FastMCP):
|
|||
),
|
||||
]
|
||||
|
||||
# Create server with custom mappings
|
||||
# Create server with custom mappings and route mapper
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=spec,
|
||||
client=httpx.AsyncClient(),
|
||||
|
|
@ -649,6 +658,8 @@ class FastMCPOpenAPI(FastMCP):
|
|||
client: httpx.AsyncClient,
|
||||
name: str | None = None,
|
||||
route_maps: list[RouteMap] | None = None,
|
||||
route_map_fn: RouteMapFn | None = None,
|
||||
mcp_component_fn: ComponentFn | None = None,
|
||||
timeout: float | None = None,
|
||||
**settings: Any,
|
||||
):
|
||||
|
|
@ -660,6 +671,12 @@ class FastMCPOpenAPI(FastMCP):
|
|||
client: httpx AsyncClient for making HTTP requests
|
||||
name: Optional name for the server
|
||||
route_maps: Optional list of RouteMap objects defining route mappings
|
||||
route_map_fn: Optional callable for advanced route type mapping.
|
||||
Receives (route, mcp_type) and returns MCPType or None.
|
||||
Called on every route, including excluded ones.
|
||||
mcp_component_fn: Optional callable for component customization.
|
||||
Receives (route, component) and can modify the component in-place.
|
||||
Called on every created component.
|
||||
timeout: Optional timeout (in seconds) for all requests
|
||||
**settings: Additional settings for FastMCP
|
||||
"""
|
||||
|
|
@ -667,6 +684,8 @@ class FastMCPOpenAPI(FastMCP):
|
|||
|
||||
self._client = client
|
||||
self._timeout = timeout
|
||||
self._route_map_fn = route_map_fn
|
||||
self._mcp_component_fn = mcp_component_fn
|
||||
|
||||
# Keep track of names to detect collisions
|
||||
self._used_names = {"tools": set(), "resources": set(), "templates": set()}
|
||||
|
|
@ -679,6 +698,22 @@ class FastMCPOpenAPI(FastMCP):
|
|||
# Determine route type based on mappings or default rules
|
||||
route_type = _determine_route_type(route, route_maps)
|
||||
|
||||
# Call route_map_fn if provided
|
||||
if self._route_map_fn is not None:
|
||||
try:
|
||||
result = self._route_map_fn(route, route_type)
|
||||
if result is not None:
|
||||
route_type = result
|
||||
logger.debug(
|
||||
f"Route {route.method} {route.path} mapping customized by route_map_fn: "
|
||||
f"type={route_type.name}"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Error in route_map_fn for {route.method} {route.path}: {e}. "
|
||||
f"Using default values."
|
||||
)
|
||||
|
||||
# Generate a default name from the route
|
||||
component_name = self._generate_default_name(route, route_type)
|
||||
|
||||
|
|
@ -688,11 +723,6 @@ class FastMCPOpenAPI(FastMCP):
|
|||
self._create_openapi_resource(route, component_name)
|
||||
elif route_type == MCPType.RESOURCE_TEMPLATE:
|
||||
self._create_openapi_template(route, component_name)
|
||||
elif route_type == MCPType.PROMPT:
|
||||
# Not implemented yet
|
||||
logger.warning(
|
||||
f"PROMPT route type not implemented: {route.method} {route.path}"
|
||||
)
|
||||
elif route_type == MCPType.EXCLUDE:
|
||||
logger.info(f"Excluding route: {route.method} {route.path}")
|
||||
|
||||
|
|
@ -795,6 +825,18 @@ class FastMCPOpenAPI(FastMCP):
|
|||
tags=set(route.tags or []),
|
||||
timeout=self._timeout,
|
||||
)
|
||||
|
||||
# Call component_fn if provided
|
||||
if self._mcp_component_fn is not None:
|
||||
try:
|
||||
self._mcp_component_fn(route, tool)
|
||||
logger.debug(f"Tool {tool_name} customized by component_fn")
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Error in component_fn for tool {tool_name}: {e}. "
|
||||
f"Using component as-is."
|
||||
)
|
||||
|
||||
# Register the tool by directly assigning to the tools dictionary
|
||||
self._tool_manager._tools[tool_name] = tool
|
||||
logger.debug(
|
||||
|
|
@ -828,6 +870,18 @@ class FastMCPOpenAPI(FastMCP):
|
|||
tags=set(route.tags or []),
|
||||
timeout=self._timeout,
|
||||
)
|
||||
|
||||
# Call component_fn if provided
|
||||
if self._mcp_component_fn is not None:
|
||||
try:
|
||||
self._mcp_component_fn(route, resource)
|
||||
logger.debug(f"Resource {resource_uri} customized by component_fn")
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Error in component_fn for resource {resource_uri}: {e}. "
|
||||
f"Using component as-is."
|
||||
)
|
||||
|
||||
# Register the resource by directly assigning to the resources dictionary
|
||||
self._resource_manager._resources[str(resource.uri)] = resource
|
||||
logger.debug(
|
||||
|
|
@ -890,6 +944,18 @@ class FastMCPOpenAPI(FastMCP):
|
|||
tags=set(route.tags or []),
|
||||
timeout=self._timeout,
|
||||
)
|
||||
|
||||
# Call component_fn if provided
|
||||
if self._mcp_component_fn is not None:
|
||||
try:
|
||||
self._mcp_component_fn(route, template)
|
||||
logger.debug(f"Template {uri_template_str} customized by component_fn")
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Error in component_fn for template {uri_template_str}: {e}. "
|
||||
f"Using component as-is."
|
||||
)
|
||||
|
||||
# Register the template by directly assigning to the templates dictionary
|
||||
self._resource_manager._templates[uri_template_str] = template
|
||||
logger.debug(
|
||||
|
|
|
|||
|
|
@ -63,7 +63,9 @@ from fastmcp.utilities.mcp_config import MCPConfig
|
|||
if TYPE_CHECKING:
|
||||
from fastmcp.client import Client
|
||||
from fastmcp.client.transports import ClientTransport
|
||||
from fastmcp.server.openapi import ComponentFn as OpenAPIComponentFn
|
||||
from fastmcp.server.openapi import FastMCPOpenAPI, RouteMap
|
||||
from fastmcp.server.openapi import RouteMapFn as OpenAPIRouteMapFn
|
||||
from fastmcp.server.proxy import FastMCPProxy
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
|
@ -1141,6 +1143,8 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
openapi_spec: dict[str, Any],
|
||||
client: httpx.AsyncClient,
|
||||
route_maps: list[RouteMap] | None = None,
|
||||
route_map_fn: OpenAPIRouteMapFn | None = None,
|
||||
mcp_component_fn: OpenAPIComponentFn | None = None,
|
||||
all_routes_as_tools: bool = False,
|
||||
**settings: Any,
|
||||
) -> FastMCPOpenAPI:
|
||||
|
|
@ -1168,6 +1172,8 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
openapi_spec=openapi_spec,
|
||||
client=client,
|
||||
route_maps=route_maps,
|
||||
route_map_fn=route_map_fn,
|
||||
mcp_component_fn=mcp_component_fn,
|
||||
**settings,
|
||||
)
|
||||
|
||||
|
|
@ -1177,6 +1183,8 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
app: Any,
|
||||
name: str | None = None,
|
||||
route_maps: list[RouteMap] | None = None,
|
||||
route_map_fn: OpenAPIRouteMapFn | None = None,
|
||||
mcp_component_fn: OpenAPIComponentFn | None = None,
|
||||
all_routes_as_tools: bool = False,
|
||||
**settings: Any,
|
||||
) -> FastMCPOpenAPI:
|
||||
|
|
@ -1212,6 +1220,8 @@ class FastMCP(Generic[LifespanResultT]):
|
|||
client=client,
|
||||
name=name,
|
||||
route_maps=route_maps,
|
||||
route_map_fn=route_map_fn,
|
||||
mcp_component_fn=mcp_component_fn,
|
||||
**settings,
|
||||
)
|
||||
|
||||
|
|
|
|||
377
tests/server/openapi/test_route_map_fn.py
Normal file
377
tests/server/openapi/test_route_map_fn.py
Normal file
|
|
@ -0,0 +1,377 @@
|
|||
"""Tests for the route_map_fn and component_fn functionality in FastMCPOpenAPI."""
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from fastmcp.server.openapi import FastMCPOpenAPI, MCPType
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_openapi_spec():
|
||||
"""Sample OpenAPI spec for testing."""
|
||||
return {
|
||||
"openapi": "3.0.0",
|
||||
"info": {"title": "Test API", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/users": {
|
||||
"get": {
|
||||
"summary": "List users",
|
||||
"operationId": "listUsers",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
"/users/{id}": {
|
||||
"get": {
|
||||
"summary": "Get user by ID",
|
||||
"operationId": "getUserById",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "id",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"schema": {"type": "string"},
|
||||
}
|
||||
],
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
"/admin/settings": {
|
||||
"get": {
|
||||
"summary": "Get admin settings",
|
||||
"operationId": "getAdminSettings",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
},
|
||||
"post": {
|
||||
"summary": "Update admin settings",
|
||||
"operationId": "updateAdminSettings",
|
||||
"requestBody": {
|
||||
"content": {"application/json": {"schema": {"type": "object"}}}
|
||||
},
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
},
|
||||
},
|
||||
"/api/data": {
|
||||
"get": {
|
||||
"summary": "Get data",
|
||||
"operationId": "getData",
|
||||
"responses": {"200": {"description": "Success"}},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def http_client():
|
||||
"""HTTP client for testing."""
|
||||
return httpx.AsyncClient()
|
||||
|
||||
|
||||
def test_route_map_fn_none(sample_openapi_spec, http_client):
|
||||
"""Test that server works correctly when route_map_fn is None."""
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
route_map_fn=None, # Explicitly set to None
|
||||
)
|
||||
|
||||
assert server.name == "Test Server"
|
||||
|
||||
|
||||
def test_route_map_fn_custom_type_conversion(sample_openapi_spec, http_client):
|
||||
"""Test that route_map_fn can convert route types."""
|
||||
|
||||
def admin_routes_to_tools(route, mcp_type):
|
||||
"""Convert all admin routes to tools."""
|
||||
if "/admin/" in route.path:
|
||||
return MCPType.TOOL
|
||||
return None
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
route_map_fn=admin_routes_to_tools,
|
||||
)
|
||||
|
||||
# Admin GET route should be converted to tool instead of resource
|
||||
tools = server._tool_manager._tools
|
||||
assert "getAdminSettings" in tools
|
||||
|
||||
# Admin POST route should still be a tool (was already)
|
||||
assert "updateAdminSettings" in tools
|
||||
|
||||
|
||||
def test_component_fn_customization(sample_openapi_spec, http_client):
|
||||
"""Test that component_fn can customize components."""
|
||||
|
||||
def customize_components(route, component):
|
||||
"""Customize components based on route."""
|
||||
from fastmcp.server.openapi import OpenAPIResource, OpenAPITool
|
||||
|
||||
# Add custom tags to all components
|
||||
component.tags.add("custom")
|
||||
|
||||
# Modify tool descriptions
|
||||
if isinstance(component, OpenAPITool):
|
||||
component.description = (component.description or "") + " [CUSTOMIZED TOOL]"
|
||||
|
||||
# Modify resource descriptions
|
||||
if isinstance(component, OpenAPIResource):
|
||||
component.description = (
|
||||
component.description or ""
|
||||
) + " [CUSTOMIZED RESOURCE]"
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
mcp_component_fn=customize_components,
|
||||
)
|
||||
|
||||
# Check that components were customized
|
||||
tools = server._tool_manager._tools
|
||||
resources = server._resource_manager._resources
|
||||
|
||||
# Tools should have custom tags and modified descriptions
|
||||
for tool in tools.values():
|
||||
assert "custom" in tool.tags
|
||||
assert "[CUSTOMIZED TOOL]" in (tool.description or "")
|
||||
|
||||
# Resources should have custom tags and modified descriptions
|
||||
for resource in resources.values():
|
||||
assert "custom" in resource.tags
|
||||
assert "[CUSTOMIZED RESOURCE]" in (resource.description or "")
|
||||
|
||||
|
||||
def test_route_map_fn_returns_none(sample_openapi_spec, http_client):
|
||||
"""Test that route_map_fn returning None uses defaults."""
|
||||
|
||||
def always_return_none(route, mcp_type):
|
||||
"""Always return None to use defaults."""
|
||||
return None
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
route_map_fn=always_return_none,
|
||||
)
|
||||
|
||||
# Should have default behavior
|
||||
assert server.name == "Test Server"
|
||||
# Check that components were created with default types
|
||||
tools = server._tool_manager._tools
|
||||
resources = server._resource_manager._resources
|
||||
templates = server._resource_manager._templates
|
||||
|
||||
# Should have tools, resources, and templates based on default mapping
|
||||
assert len(tools) > 0
|
||||
assert len(resources) > 0
|
||||
assert len(templates) > 0
|
||||
|
||||
|
||||
def test_route_map_fn_called_for_excluded_routes(sample_openapi_spec, http_client):
|
||||
"""Test that route_map_fn is called for excluded routes and can rescue them."""
|
||||
|
||||
from fastmcp.server.openapi import RouteMap
|
||||
|
||||
# Exclude all admin routes
|
||||
route_maps = [
|
||||
RouteMap(
|
||||
methods=["GET", "POST"], pattern=r".*/admin/.*", mcp_type=MCPType.EXCLUDE
|
||||
)
|
||||
]
|
||||
|
||||
called_routes = []
|
||||
|
||||
def track_calls_and_rescue(route, mcp_type):
|
||||
"""Track which routes the function is called for and rescue some excluded routes."""
|
||||
called_routes.append(route.path)
|
||||
|
||||
# Rescue the admin GET route by converting it to a tool
|
||||
if route.path == "/admin/settings" and route.method == "GET":
|
||||
return MCPType.TOOL
|
||||
|
||||
return None # Accept the assignment for other routes
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
route_maps=route_maps,
|
||||
route_map_fn=track_calls_and_rescue,
|
||||
)
|
||||
|
||||
# route_map_fn should now be called for all routes, including excluded admin routes
|
||||
assert "/admin/settings" in called_routes
|
||||
assert "/users" in called_routes
|
||||
assert "/users/{id}" in called_routes
|
||||
assert "/api/data" in called_routes
|
||||
|
||||
# The rescued admin GET route should now be a tool
|
||||
tools = server._tool_manager._tools
|
||||
assert "getAdminSettings" in tools
|
||||
|
||||
# The admin POST route should still be excluded (not rescued)
|
||||
assert "updateAdminSettings" not in tools
|
||||
|
||||
|
||||
def test_route_map_fn_error_handling(sample_openapi_spec, http_client):
|
||||
"""Test that errors in route_map_fn are handled gracefully."""
|
||||
|
||||
def error_function(route, mcp_type):
|
||||
"""Function that raises an error."""
|
||||
if route.path == "/users":
|
||||
raise ValueError("Test error")
|
||||
return None
|
||||
|
||||
# Should not raise an error, but log a warning
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
route_map_fn=error_function,
|
||||
)
|
||||
|
||||
# Server should still be created successfully
|
||||
assert server.name == "Test Server"
|
||||
|
||||
|
||||
def test_component_fn_error_handling(sample_openapi_spec, http_client):
|
||||
"""Test that errors in component_fn are handled gracefully."""
|
||||
|
||||
def error_function(route, component):
|
||||
"""Function that raises an error."""
|
||||
if route.path == "/users":
|
||||
raise ValueError("Test error in component_fn")
|
||||
|
||||
# Should not raise an error, but log a warning
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
mcp_component_fn=error_function,
|
||||
)
|
||||
|
||||
# Server should still be created successfully
|
||||
assert server.name == "Test Server"
|
||||
|
||||
|
||||
def test_combined_route_map_fn_and_component_fn(sample_openapi_spec, http_client):
|
||||
"""Test using both route_map_fn and component_fn together."""
|
||||
|
||||
def route_mapper(route, mcp_type):
|
||||
"""Convert admin routes to tools."""
|
||||
if "/admin/" in route.path:
|
||||
return MCPType.TOOL
|
||||
return None
|
||||
|
||||
def component_customizer(route, component):
|
||||
"""Add admin tag to admin components."""
|
||||
if "/admin/" in route.path:
|
||||
component.tags.add("admin")
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
route_map_fn=route_mapper,
|
||||
mcp_component_fn=component_customizer,
|
||||
)
|
||||
|
||||
# Check that both functions worked
|
||||
tools = server._tool_manager._tools
|
||||
|
||||
# Admin GET route should be converted to tool
|
||||
assert "getAdminSettings" in tools
|
||||
admin_tool = tools["getAdminSettings"]
|
||||
assert "admin" in admin_tool.tags
|
||||
|
||||
# Admin POST route should have admin tag
|
||||
admin_post_tool = tools["updateAdminSettings"]
|
||||
assert "admin" in admin_post_tool.tags
|
||||
|
||||
|
||||
def test_route_map_fn_signature_validation():
|
||||
"""Test that route_map_fn has the correct signature."""
|
||||
from fastmcp.server.openapi import RouteMapFn
|
||||
from fastmcp.utilities import openapi
|
||||
|
||||
# This is more of a type checking test
|
||||
def valid_route_map_fn(
|
||||
route: openapi.HTTPRoute, mcp_type: MCPType
|
||||
) -> MCPType | None:
|
||||
return None
|
||||
|
||||
# Should be assignable to RouteMapFn type
|
||||
fn: RouteMapFn = valid_route_map_fn
|
||||
assert callable(fn)
|
||||
|
||||
|
||||
def test_component_fn_signature_validation():
|
||||
"""Test that component_fn has the correct signature."""
|
||||
from fastmcp.server.openapi import (
|
||||
ComponentFn,
|
||||
OpenAPIResource,
|
||||
OpenAPIResourceTemplate,
|
||||
OpenAPITool,
|
||||
)
|
||||
from fastmcp.utilities import openapi
|
||||
|
||||
# This is more of a type checking test
|
||||
def valid_component_fn(
|
||||
route: openapi.HTTPRoute,
|
||||
component: OpenAPITool | OpenAPIResource | OpenAPIResourceTemplate,
|
||||
) -> None:
|
||||
pass
|
||||
|
||||
# Should be assignable to ComponentFn type
|
||||
fn: ComponentFn = valid_component_fn
|
||||
assert callable(fn)
|
||||
|
||||
|
||||
def test_route_map_fn_can_rescue_excluded_routes(sample_openapi_spec, http_client):
|
||||
"""Test that route_map_fn can rescue routes that were excluded by RouteMap."""
|
||||
|
||||
from fastmcp.server.openapi import RouteMap
|
||||
|
||||
# Exclude ALL routes by default
|
||||
route_maps = [
|
||||
RouteMap(mcp_type=MCPType.EXCLUDE) # Catch-all exclusion
|
||||
]
|
||||
|
||||
def rescue_users_routes(route, mcp_type):
|
||||
"""Rescue only user-related routes."""
|
||||
if "/users" in route.path:
|
||||
# Rescue user routes as tools
|
||||
return MCPType.TOOL
|
||||
# Let everything else stay excluded
|
||||
return None
|
||||
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=sample_openapi_spec,
|
||||
client=http_client,
|
||||
name="Test Server",
|
||||
route_maps=route_maps,
|
||||
route_map_fn=rescue_users_routes,
|
||||
)
|
||||
|
||||
# Only user routes should be rescued as tools
|
||||
tools = server._tool_manager._tools
|
||||
resources = server._resource_manager._resources
|
||||
templates = server._resource_manager._templates
|
||||
|
||||
# Should have user-related tools
|
||||
assert "listUsers" in tools
|
||||
assert "getUserById" in tools
|
||||
|
||||
# Should have no resources or templates (everything excluded except rescued tools)
|
||||
assert len(resources) == 0
|
||||
assert len(templates) == 0
|
||||
|
||||
# Admin and API routes should still be excluded
|
||||
assert "getAdminSettings" not in tools
|
||||
assert "updateAdminSettings" not in tools
|
||||
assert "getData" not in tools
|
||||
Loading…
Add table
Add a link
Reference in a new issue