Merge pull request #578 from jlowin/headers

Permit more flexible name generation for OpenAPI servers
This commit is contained in:
Jeremiah Lowin 2025-05-23 18:02:46 -04:00 committed by GitHub
commit 2ab07cbcd1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 423 additions and 99 deletions

View file

@ -231,6 +231,38 @@ mcp = FastMCP.from_openapi(
## Customizing MCP Components
### Component Names
<VersionBadge version="2.5.0" />
FastMCP automatically generates names for MCP components based on the OpenAPI specification. By default, it uses the `operationId` from your OpenAPI spec, up to the first double underscore (`__`).
All component names are automatically:
- **Slugified**: Spaces and special characters are converted to underscores or removed
- **Truncated**: Limited to 56 characters maximum to ensure compatibility
- **Unique**: If multiple components have the same name, a number is automatically appended to make them unique
For more control over component names, you can provide an `mcp_names` dictionary that maps `operationId` values to your desired names. The `operationId` must be exactly as it appears in the OpenAPI spec. The provided name will always be slugified and truncated.
```python {5-9}
from fastmcp import FastMCP
mcp = FastMCP.from_openapi(
...
mcp_names={
"list_users__with_pagination": "user_list",
"create_user__admin_required": "create_user",
"get_user_details__admin_required": "user_detail",
}
)
```
Any `operationId` not found in `mcp_names` will use the default strategy (operationId up to the first `__`).
### Advanced Customization
<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.
@ -271,7 +303,6 @@ mcp = FastMCP.from_openapi(
mcp_component_fn=customize_components,
)
```
## Request Parameter Handling
FastMCP intelligently handles different types of parameters in OpenAPI requests:
@ -376,15 +407,15 @@ from fastmcp import FastMCP
# Your FastAPI app
app = FastAPI(title="My API", version="1.0.0")
@app.get("/items", tags=["items"])
@app.get("/items", tags=["items"], operation_id="list_items")
def list_items():
return [{"id": 1, "name": "Item 1"}, {"id": 2, "name": "Item 2"}]
@app.get("/items/{item_id}", tags=["items", "detail"])
@app.get("/items/{item_id}", tags=["items", "detail"], operation_id="get_item")
def get_item(item_id: int):
return {"id": item_id, "name": f"Item {item_id}"}
@app.post("/items", tags=["items", "create"])
@app.post("/items", tags=["items", "create"], operation_id="create_item")
def create_item(name: str):
return {"id": 3, "name": name}
@ -395,6 +426,8 @@ if __name__ == "__main__":
mcp.run() # Run as MCP server
```
Note that operation ids are optional, but are used to create component names. You can also provide custom names, just like with OpenAPI specs.
<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>
@ -413,6 +446,7 @@ mcp = FastMCP.from_fastapi(
app=app,
name="My Custom Server",
timeout=5.0,
mcp_names={"operationId": "friendly_name"}, # Custom component names
route_maps=[
# Admin endpoints become tools
RouteMap(methods="*", pattern=r"^/admin/.*", mcp_type=MCPType.TOOL),
@ -421,6 +455,9 @@ mcp = FastMCP.from_fastapi(
],
route_map_fn=my_route_mapper,
mcp_component_fn=my_component_customizer,
mcp_names={
"get_user_details_users__user_id__get": "get_user_details",
}
)
```
@ -430,4 +467,3 @@ mcp = FastMCP.from_fastapi(
- **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

View file

@ -6,6 +6,7 @@ import enum
import json
import re
import warnings
from collections import Counter
from collections.abc import Callable
from dataclasses import dataclass, field
from re import Pattern
@ -36,6 +37,29 @@ logger = get_logger(__name__)
HttpMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "HEAD"]
def _slugify(text: str) -> str:
"""
Convert text to a URL-friendly slug format that only contains lowercase
letters, uppercase letters, numbers, and underscores.
"""
if not text:
return ""
# Replace spaces and common separators with underscores
slug = re.sub(r"[\s\-\.]+", "_", text)
# Remove non-alphanumeric characters except underscores
slug = re.sub(r"[^a-zA-Z0-9_]", "", slug)
# Remove multiple consecutive underscores
slug = re.sub(r"_+", "_", slug)
# Remove leading/trailing underscores
slug = slug.strip("_")
return slug
def _get_mcp_client_headers() -> dict[str, str]:
"""
Extract headers from the current MCP client HTTP request if available.
@ -695,6 +719,7 @@ class FastMCPOpenAPI(FastMCP):
route_maps: list[RouteMap] | None = None,
route_map_fn: RouteMapFn | None = None,
mcp_component_fn: ComponentFn | None = None,
mcp_names: dict[str, str] | None = None,
timeout: float | None = None,
**settings: Any,
):
@ -712,6 +737,11 @@ class FastMCPOpenAPI(FastMCP):
mcp_component_fn: Optional callable for component customization.
Receives (route, component) and can modify the component in-place.
Called on every created component.
mcp_names: Optional dictionary mapping operationId to desired component names.
If an operationId is not in the dictionary, falls back to using the
operationId up to the first double underscore. If no operationId exists,
falls back to slugified summary or path-based naming.
All names are truncated to 56 characters maximum.
timeout: Optional timeout (in seconds) for all requests
**settings: Additional settings for FastMCP
"""
@ -721,9 +751,15 @@ class FastMCPOpenAPI(FastMCP):
self._timeout = timeout
self._route_map_fn = route_map_fn
self._mcp_component_fn = mcp_component_fn
self._mcp_names = mcp_names or {}
# Keep track of names to detect collisions
self._used_names = {"tools": set(), "resources": set(), "templates": set()}
self._used_names = {
"tool": Counter(),
"resource": Counter(),
"resource_template": Counter(),
"prompt": Counter(),
}
http_routes = openapi.parse_openapi_to_http_routes(openapi_spec)
@ -766,40 +802,31 @@ class FastMCPOpenAPI(FastMCP):
def _generate_default_name(
self, route: openapi.HTTPRoute, mcp_type: MCPType
) -> str:
"""Generate a default name from the route path."""
# First check for OpenAPI operationId which takes precedence
"""Generate a default name from the route using the configured strategy."""
name = ""
# First check if there's a custom mapping for this operationId
if route.operation_id:
return route.operation_id
# For path-based naming, clean up the path
path_parts = route.path.strip("/").split("/")
# Remove path parameters (parts with {})
clean_parts = []
for part in path_parts:
if part.startswith("{") and part.endswith("}"):
# For templates, include parameter name without braces
if mcp_type == MCPType.RESOURCE_TEMPLATE:
param_name = part[1:-1] # Remove braces
clean_parts.append(param_name)
if route.operation_id in self._mcp_names:
name = self._mcp_names[route.operation_id]
else:
clean_parts.append(part)
# If there's a double underscore in the operationId, use the first part
name = route.operation_id.split("__")[0]
else:
name = route.summary or f"{route.method}_{route.path}"
# Join the parts
resource_name = "_".join(clean_parts)
name = _slugify(name)
# For tools, might be useful to keep the method for clarity on what it does
if mcp_type == MCPType.TOOL:
# Only include method if it helps distinguish (POST, PUT, PATCH, DELETE)
# For GET we don't need the method as it's implied for resources
if route.method != "GET":
resource_name = f"{route.method.lower()}_{resource_name}"
# Truncate to 56 characters maximum
if len(name) > 56:
name = name[:56]
return resource_name
return name
def _get_unique_name(
self, name: str, component_type: Literal["tools", "resources", "templates"]
self,
name: str,
component_type: Literal["tool", "resource", "resource_template", "prompt"],
) -> str:
"""
Ensure the name is unique within its component type by appending numbers if needed.
@ -812,23 +839,18 @@ class FastMCPOpenAPI(FastMCP):
str: A unique name for the component
"""
# Check if the name is already used
if name not in self._used_names[component_type]:
self._used_names[component_type].add(name)
self._used_names[component_type][name] += 1
if self._used_names[component_type][name] == 1:
return name
# Find the next available number suffix
counter = 2
while f"{name}_{counter}" in self._used_names[component_type]:
counter += 1
else:
# Create the new name
new_name = f"{name}_{self._used_names[component_type][name]}"
logger.debug(
f"Name collision detected: '{name}' already exists as a {component_type[:-1]}. "
f"Using '{new_name}' instead."
)
# Create the new name
new_name = f"{name}_{counter}"
logger.debug(
f"Name collision detected: '{name}' already exists as a {component_type[:-1]}. "
f"Using '{new_name}' instead."
)
self._used_names[component_type].add(new_name)
return new_name
def _create_openapi_tool(self, route: openapi.HTTPRoute, name: str):
@ -836,7 +858,7 @@ class FastMCPOpenAPI(FastMCP):
combined_schema = _combine_schemas(route)
# Get a unique tool name
tool_name = self._get_unique_name(name, "tools")
tool_name = self._get_unique_name(name, "tool")
base_description = (
route.description
@ -882,7 +904,7 @@ class FastMCPOpenAPI(FastMCP):
def _create_openapi_resource(self, route: openapi.HTTPRoute, name: str):
"""Creates and registers an OpenAPIResource with enhanced description."""
# Get a unique resource name
resource_name = self._get_unique_name(name, "resources")
resource_name = self._get_unique_name(name, "resource")
resource_uri = f"resource://{resource_name}"
base_description = (
@ -927,7 +949,7 @@ class FastMCPOpenAPI(FastMCP):
def _create_openapi_template(self, route: openapi.HTTPRoute, name: str):
"""Creates and registers an OpenAPIResourceTemplate with enhanced description."""
# Get a unique template name
template_name = self._get_unique_name(name, "templates")
template_name = self._get_unique_name(name, "resource_template")
path_params = [p.name for p in route.parameters if p.location == "path"]
path_params.sort() # Sort for consistent URIs

View file

@ -1170,6 +1170,7 @@ class FastMCP(Generic[LifespanResultT]):
route_maps: list[RouteMap] | None = None,
route_map_fn: OpenAPIRouteMapFn | None = None,
mcp_component_fn: OpenAPIComponentFn | None = None,
mcp_names: dict[str, str] | None = None,
all_routes_as_tools: bool = False,
**settings: Any,
) -> FastMCPOpenAPI:
@ -1199,6 +1200,7 @@ class FastMCP(Generic[LifespanResultT]):
route_maps=route_maps,
route_map_fn=route_map_fn,
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
**settings,
)
@ -1210,6 +1212,7 @@ class FastMCP(Generic[LifespanResultT]):
route_maps: list[RouteMap] | None = None,
route_map_fn: OpenAPIRouteMapFn | None = None,
mcp_component_fn: OpenAPIComponentFn | None = None,
mcp_names: dict[str, str] | None = None,
all_routes_as_tools: bool = False,
httpx_client_kwargs: dict[str, Any] | None = None,
**settings: Any,
@ -1253,6 +1256,7 @@ class FastMCP(Generic[LifespanResultT]):
route_maps=route_maps,
route_map_fn=route_map_fn,
mcp_component_fn=mcp_component_fn,
mcp_names=mcp_names,
**settings,
)

View file

@ -130,7 +130,7 @@ class TestClientHeaders:
transport=SSETransport(sse_server, headers={"X-TEST": "test-123"})
) as client:
result = await client.read_resource(
"resource://get_header_by_name_headers__header_name__get/x-test"
"resource://get_header_by_name_headers/x-test"
)
assert isinstance(result[0], TextResourceContents)
header = json.loads(result[0].text)
@ -143,7 +143,7 @@ class TestClientHeaders:
)
) as client:
result = await client.read_resource(
"resource://get_header_by_name_headers__header_name__get/x-test"
"resource://get_header_by_name_headers/x-test"
)
assert isinstance(result[0], TextResourceContents)
header = json.loads(result[0].text)

View file

@ -208,7 +208,7 @@ class TestTools:
},
)
assert tools[1].model_dump() == dict(
name="update_user_name_users__user_id__name_patch",
name="update_user_name_users",
annotations=None,
description=IsStr(
regex=r"^Update a user's name\..*$", regex_flags=re.DOTALL
@ -248,9 +248,7 @@ class TestTools:
# Check that the user was created via MCP
async with Client(fastmcp_openapi_server) as client:
user_response = await client.read_resource(
"resource://get_user_users__user_id__get/4"
)
user_response = await client.read_resource("resource://get_user_users/4")
assert isinstance(user_response[0], TextResourceContents)
response_text = user_response[0].text
user = json.loads(response_text)
@ -264,7 +262,7 @@ class TestTools:
"""
async with Client(fastmcp_openapi_server) as client:
tool_response = await client.call_tool(
"update_user_name_users__user_id__name_patch",
"update_user_name_users",
{"user_id": 1, "name": "XYZ"},
)
@ -282,9 +280,7 @@ class TestTools:
# Check that the user was updated via MCP
async with Client(fastmcp_openapi_server) as client:
user_response = await client.read_resource(
"resource://get_user_users__user_id__get/1"
)
user_response = await client.read_resource("resource://get_user_users/1")
assert isinstance(user_response[0], TextResourceContents)
response_text = user_response[0].text
user = json.loads(response_text)
@ -387,18 +383,14 @@ class TestResourceTemplates:
async with Client(fastmcp_openapi_server) as client:
resource_templates = await client.list_resource_templates()
assert len(resource_templates) == 2
assert resource_templates[0].name == "get_user_users__user_id__get"
assert resource_templates[0].name == "get_user_users"
assert (
resource_templates[0].uriTemplate
== r"resource://get_user_users__user_id__get/{user_id}"
)
assert (
resource_templates[1].name
== "get_user_active_state_users__user_id___is_active__get"
resource_templates[0].uriTemplate == r"resource://get_user_users/{user_id}"
)
assert resource_templates[1].name == "get_user_active_state_users"
assert (
resource_templates[1].uriTemplate
== r"resource://get_user_active_state_users__user_id___is_active__get/{is_active}/{user_id}"
== r"resource://get_user_active_state_users/{is_active}/{user_id}"
)
async def test_get_resource_template(
@ -413,7 +405,7 @@ class TestResourceTemplates:
user_id = 2
async with Client(fastmcp_openapi_server) as client:
resource_response = await client.read_resource(
f"resource://get_user_users__user_id__get/{user_id}"
f"resource://get_user_users/{user_id}"
)
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
@ -436,7 +428,7 @@ class TestResourceTemplates:
is_active = True
async with Client(fastmcp_openapi_server) as client:
resource_response = await client.read_resource(
f"resource://get_user_active_state_users__user_id___is_active__get/{is_active}/{user_id}"
f"resource://get_user_active_state_users/{is_active}/{user_id}"
)
assert isinstance(resource_response[0], TextResourceContents)
response_text = resource_response[0].text
@ -472,11 +464,7 @@ class TestTagTransfer:
(t for t in tools if t.name == "create_user_users_post"), None
)
update_user_tool = next(
(
t
for t in tools
if t.name == "update_user_name_users__user_id__name_patch"
),
(t for t in tools if t.name == "update_user_name_users"),
None,
)
@ -524,7 +512,7 @@ class TestTagTransfer:
# Find the get_user template
get_user_template = next(
(t for t in templates if t.name == "get_user_users__user_id__get"), None
(t for t in templates if t.name == "get_user_users"), None
)
assert get_user_template is not None
@ -545,7 +533,7 @@ class TestTagTransfer:
# Find the get_user template
get_user_template = next(
(t for t in templates if t.name == "get_user_users__user_id__get"), None
(t for t in templates if t.name == "get_user_users"), None
)
assert get_user_template is not None
@ -553,7 +541,7 @@ class TestTagTransfer:
# Manually create a resource from template
params = {"user_id": 1}
resource = await get_user_template.create_resource(
"resource://get_user_users__user_id__get/1", params
"resource://get_user_users/1", params
)
# Verify tags are preserved from template to resource
@ -997,7 +985,7 @@ async def test_none_path_parameters_rejected(
# get_user has a required path parameter user_id
with pytest.raises(ToolError, match="Missing required path parameters"):
await client.call_tool(
"update_user_name_users__user_id__name_patch",
"update_user_name_users",
{
"user_id": None, # This should cause an error
"name": "New Name",
@ -1560,9 +1548,7 @@ class TestFastAPIDescriptionPropagation:
async def test_template_includes_function_docstring(self, fastapi_server):
"""Test that a ResourceTemplate includes the function docstring."""
templates = list(fastapi_server._resource_manager.get_templates().values())
get_template = next(
(t for t in templates if "items__item_id__get" in t.name), None
)
get_template = next((t for t in templates if "get_item_items" in t.name), None)
assert get_template is not None, "GET /items/{item_id} template wasn't created"
description = get_template.description or ""
@ -1577,9 +1563,7 @@ class TestFastAPIDescriptionPropagation:
are not properly propagated to the OpenAPI schema. The parameters appear but without the description.
"""
templates = list(fastapi_server._resource_manager.get_templates().values())
get_template = next(
(t for t in templates if "items__item_id__get" in t.name), None
)
get_template = next((t for t in templates if "get_item_items" in t.name), None)
assert get_template is not None, "GET /items/{item_id} template wasn't created"
description = get_template.description or ""
@ -1599,9 +1583,7 @@ class TestFastAPIDescriptionPropagation:
are not properly propagated to the OpenAPI schema. The parameters appear but without the description.
"""
templates = list(fastapi_server._resource_manager.get_templates().values())
get_template = next(
(t for t in templates if "items__item_id__get" in t.name), None
)
get_template = next((t for t in templates if "get_item_items" in t.name), None)
assert get_template is not None, "GET /items/{item_id} template wasn't created"
description = get_template.description or ""
@ -1617,9 +1599,7 @@ class TestFastAPIDescriptionPropagation:
async def test_template_parameter_schema_includes_description(self, fastapi_server):
"""Test that a ResourceTemplate's parameter schema includes parameter descriptions."""
templates = list(fastapi_server._resource_manager.get_templates().values())
get_template = next(
(t for t in templates if "items__item_id__get" in t.name), None
)
get_template = next((t for t in templates if "get_item_items" in t.name), None)
assert get_template is not None, "GET /items/{item_id} template wasn't created"
assert "properties" in get_template.parameters, (
@ -1691,7 +1671,7 @@ class TestFastAPIDescriptionPropagation:
async with Client(fastapi_server) as client:
templates = await client.list_resource_templates()
get_template = next(
(t for t in templates if "items__item_id__get" in t.name), None
(t for t in templates if "get_item_items" in t.name), None
)
assert get_template is not None, (
@ -1821,9 +1801,7 @@ class TestEnumHandling:
tools = server._tool_manager.list_tools()
# Find the read_item tool
read_item_tool = next(
(t for t in tools if t.name == "read_item_items__item_id__post"), None
)
read_item_tool = next((t for t in tools if t.name == "read_item_items"), None)
# Verify the tool exists
assert read_item_tool is not None, "read_item tool wasn't created"
@ -2136,3 +2114,287 @@ class TestRouteMapTags:
"getMetrics",
}
assert tool_names == expected_tools
class TestMCPNames:
"""Tests for the mcp_names dictionary functionality."""
@pytest.fixture
def mcp_names_openapi_spec(self) -> dict:
"""OpenAPI spec with various operationIds for testing naming strategies."""
return {
"openapi": "3.1.0",
"info": {"title": "MCP Names Test API", "version": "1.0.0"},
"paths": {
"/users": {
"get": {
"operationId": "list_users__with_pagination",
"summary": "Get All Users",
"responses": {"200": {"description": "Success"}},
},
"post": {
"operationId": "create_user_admin__special_permissions",
"summary": "Create New User",
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {"name": {"type": "string"}},
"required": ["name"],
}
}
},
},
"responses": {"201": {"description": "Created"}},
},
},
"/users/{id}": {
"get": {
"operationId": "get_user_by_id__admin_only",
"summary": "Fetch Single User Profile",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "integer"},
}
],
"responses": {"200": {"description": "Success"}},
}
},
"/very-long-endpoint-name": {
"get": {
"operationId": "this_is_a_very_long_operation_id_that_exceeds_fifty_six_characters_and_should_be_truncated",
"summary": "This Is A Very Long Summary That Should Also Be Truncated When Used As Name",
"responses": {"200": {"description": "Success"}},
}
},
"/special": {
"get": {
"operationId": "special-chars@and#spaces in$operation%id",
"summary": "Special Chars & Spaces In Summary!",
"responses": {"200": {"description": "Success"}},
}
},
},
}
@pytest.fixture
async def mock_client(self) -> httpx.AsyncClient:
"""Mock client for testing."""
async def _responder(request):
return httpx.Response(200, json={"status": "ok"})
transport = httpx.MockTransport(_responder)
return httpx.AsyncClient(transport=transport, base_url="http://test")
async def test_mcp_names_custom_mapping(self, mcp_names_openapi_spec, mock_client):
"""Test that mcp_names dictionary provides custom names for components."""
mcp_names = {
"list_users__with_pagination": "user_list",
"create_user_admin__special_permissions": "admin_create_user",
"get_user_by_id__admin_only": "user_detail",
}
server = FastMCPOpenAPI(
openapi_spec=mcp_names_openapi_spec,
client=mock_client,
mcp_names=mcp_names,
)
# Check tools use custom names
tools = server._tool_manager.list_tools()
tool_names = {tool.name for tool in tools}
assert "admin_create_user" in tool_names
# Check resource templates use custom names
templates = list(server._resource_manager.get_templates().values())
template_names = {template.name for template in templates}
assert "user_detail" in template_names
# Check resources use custom names
resources = list(server._resource_manager.get_resources().values())
resource_names = {resource.name for resource in resources}
assert "user_list" in resource_names
async def test_mcp_names_fallback_to_operation_id_short(
self, mcp_names_openapi_spec, mock_client
):
"""Test fallback to operationId up to double underscore when not in mcp_names."""
# Only provide mapping for one operationId
mcp_names = {
"list_users__with_pagination": "custom_user_list",
}
server = FastMCPOpenAPI(
openapi_spec=mcp_names_openapi_spec,
client=mock_client,
mcp_names=mcp_names,
)
tools = server._tool_manager.list_tools()
tool_names = {tool.name for tool in tools}
templates = list(server._resource_manager.get_templates().values())
template_names = {template.name for template in templates}
resources = list(server._resource_manager.get_resources().values())
resource_names = {resource.name for resource in resources}
# Custom mapped name should be used
assert "custom_user_list" in resource_names
# Unmapped operationIds should use short version (up to __)
assert "create_user_admin" in tool_names
assert "get_user_by_id" in template_names
async def test_names_are_slugified(self, mcp_names_openapi_spec, mock_client):
"""Test that names are properly slugified (spaces, special chars removed)."""
server = FastMCPOpenAPI(
openapi_spec=mcp_names_openapi_spec,
client=mock_client,
)
resources = list(server._resource_manager.get_resources().values())
resource_names = {
resource.name for resource in resources if resource.name is not None
}
# Special chars and spaces should be slugified
slugified_name = next(
(name for name in resource_names if "special" in name), None
)
assert slugified_name is not None
# Should not contain special characters or spaces
assert "@" not in slugified_name
assert "#" not in slugified_name
assert "$" not in slugified_name
assert "%" not in slugified_name
assert " " not in slugified_name
async def test_names_are_truncated_to_56_chars(
self, mcp_names_openapi_spec, mock_client
):
"""Test that names are truncated to 56 characters maximum."""
server = FastMCPOpenAPI(
openapi_spec=mcp_names_openapi_spec,
client=mock_client,
)
# Check all component types
all_names = []
tools = server._tool_manager.list_tools()
all_names.extend(tool.name for tool in tools)
resources = list(server._resource_manager.get_resources().values())
all_names.extend(resource.name for resource in resources)
templates = list(server._resource_manager.get_templates().values())
all_names.extend(template.name for template in templates)
# All names should be 56 characters or less
for name in all_names:
assert len(name) <= 56, (
f"Name '{name}' exceeds 56 characters (length: {len(name)})"
)
# Verify that the long operationId was actually truncated
long_name = next((name for name in all_names if len(name) > 50), None)
assert long_name is not None, "Expected to find a truncated name for testing"
async def test_mcp_names_with_from_openapi_classmethod(
self, mcp_names_openapi_spec, mock_client
):
"""Test mcp_names works with FastMCP.from_openapi() classmethod."""
mcp_names = {
"list_users__with_pagination": "openapi_user_list",
}
server = FastMCP.from_openapi(
openapi_spec=mcp_names_openapi_spec,
client=mock_client,
mcp_names=mcp_names,
)
resources = list(server._resource_manager.get_resources().values())
resource_names = {resource.name for resource in resources}
assert "openapi_user_list" in resource_names
async def test_mcp_names_with_from_fastapi_classmethod(self):
"""Test mcp_names works with FastMCP.from_fastapi() classmethod."""
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="FastAPI MCP Names Test")
class User(BaseModel):
name: str
@app.get("/users", operation_id="list_users__with_filters")
async def get_users() -> list[User]:
return [User(name="test")]
@app.post("/users", operation_id="create_user__admin_required")
async def create_user(user: User) -> User:
return user
mcp_names = {
"list_users__with_filters": "fastapi_user_list",
"create_user__admin_required": "fastapi_create_user",
}
server = FastMCP.from_fastapi(
app=app,
mcp_names=mcp_names,
)
tools = server._tool_manager.list_tools()
tool_names = {tool.name for tool in tools}
resources = list(server._resource_manager.get_resources().values())
resource_names = {resource.name for resource in resources}
assert "fastapi_create_user" in tool_names
assert "fastapi_user_list" in resource_names
async def test_mcp_names_custom_names_are_also_truncated(
self, mcp_names_openapi_spec, mock_client
):
"""Test that custom names in mcp_names are also truncated to 56 characters."""
# Provide a custom name that's longer than 56 characters
very_long_custom_name = "this_is_a_very_long_custom_name_that_exceeds_fifty_six_characters_and_should_be_truncated"
mcp_names = {
"list_users__with_pagination": very_long_custom_name,
}
server = FastMCPOpenAPI(
openapi_spec=mcp_names_openapi_spec,
client=mock_client,
mcp_names=mcp_names,
)
resources = list(server._resource_manager.get_resources().values())
resource_names = {
resource.name for resource in resources if resource.name is not None
}
# Find the resource that should have the custom name
truncated_name = next(
(
name
for name in resource_names
if "this_is_a_very_long_custom_name" in name
),
None,
)
assert truncated_name is not None
assert len(truncated_name) <= 56
assert (
len(truncated_name) == 56
) # Should be exactly 56 since original was longer

View file

@ -84,7 +84,7 @@ async def test_fastmcp_from_openapi(array_path_spec, mock_client):
# Verify the tool was created using the MCP protocol method
tools_result = await mcp.get_tools()
tool_names = [tool.name for tool in tools_result.values()]
assert "test-operation" in tool_names
assert "test_operation" in tool_names
async def test_array_path_parameter_handling(mock_client):
@ -93,7 +93,7 @@ async def test_array_path_parameter_handling(mock_client):
route = HTTPRoute(
path="/select/{days}",
method="PUT",
operation_id="test-operation",
operation_id="test_operation",
parameters=[
ParameterInfo(
name="days",
@ -122,7 +122,7 @@ async def test_array_path_parameter_handling(mock_client):
tool = OpenAPITool(
client=mock_client,
route=route,
name="test-operation",
name="test_operation",
description="Test operation",
parameters={},
)
@ -163,7 +163,7 @@ async def test_integration_array_path_parameter(array_path_spec, mock_client):
mcp = FastMCP.from_openapi(array_path_spec, client=mock_client)
# Call the tool with a single value
await mcp._mcp_call_tool("test-operation", {"days": ["monday"]})
await mcp._mcp_call_tool("test_operation", {"days": ["monday"]})
# Check the request was made correctly
mock_client.request.assert_called_with(
@ -177,7 +177,7 @@ async def test_integration_array_path_parameter(array_path_spec, mock_client):
mock_client.request.reset_mock()
# Call the tool with multiple values
await mcp._mcp_call_tool("test-operation", {"days": ["monday", "tuesday"]})
await mcp._mcp_call_tool("test_operation", {"days": ["monday", "tuesday"]})
# Check the request was made correctly
mock_client.request.assert_called_with(