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

Merge pull request #124 from jlowin/openapi-tags

Browse source 
Ensure that openapi tags are transferred to MCP objects
This commit is contained in:
Jeremiah Lowin 2025-04-12 12:42:35 -04:00 • committed by GitHub
commit 6e7459e7fd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 261 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/fastmcp/server/openapi.py
View file

@ -115,6 +115,7 @@ class OpenAPITool(Tool):
parameters: dict[str, Any],
fn_metadata: Any,
is_async: bool = True,
tags: set[str] = set(),
):
super().__init__(
name=name,
@ -124,6 +125,7 @@ class OpenAPITool(Tool):
fn_metadata=fn_metadata,
is_async=is_async,
context_kwarg="context", # Default context keyword argument
tags=tags,
)
self._client = client
self._route = route
@ -242,12 +244,14 @@ class OpenAPIResource(Resource):
name: str,
description: str,
mime_type: str = "application/json",
tags: set[str] = set(),
):
super().__init__(
uri=AnyUrl(uri), # Convert string to AnyUrl
name=name,
description=description,
mime_type=mime_type,
tags=tags,
)
self._client = client
self._route = route
@ -332,6 +336,7 @@ class OpenAPIResourceTemplate(ResourceTemplate):
name: str,
description: str,
parameters: dict[str, Any],
tags: set[str] = set(),
):
super().__init__(
uri_template=uri_template,
@ -339,6 +344,7 @@ class OpenAPIResourceTemplate(ResourceTemplate):
description=description,
fn=self._create_resource_fn,
parameters=parameters,
tags=tags,
)
self._client = client
self._route = route
@ -405,6 +411,7 @@ class OpenAPIResourceTemplate(ResourceTemplate):
description=self.description
or f"Resource for {self._route.path}", # Provide default if None
mime_type="application/json", # Default, will be updated when read
tags=set(self._route.tags or []),
)
@ -525,10 +532,13 @@ class FastMCPOpenAPI(FastMCP):
parameters=combined_schema,
fn_metadata=func_metadata(_openapi_passthrough),
is_async=True,
tags=set(route.tags or []),
)
# Register the tool by directly assigning to the tools dictionary
self._tool_manager._tools[tool_name] = tool
logger.debug(f"Registered TOOL: {tool_name} ({route.method} {route.path})")
logger.debug(
f"Registered TOOL: {tool_name} ({route.method} {route.path}) with tags: {route.tags}"
)
def _create_openapi_resource(self, route: openapi.HTTPRoute, operation_id: str):
"""Creates and registers an OpenAPIResource with enhanced description."""
@ -550,11 +560,12 @@ class FastMCPOpenAPI(FastMCP):
uri=resource_uri,
name=resource_name,
description=enhanced_description,
tags=set(route.tags or []),
)
# Register the resource by directly assigning to the resources dictionary
self._resource_manager._resources[str(resource.uri)] = resource
logger.debug(
f"Registered RESOURCE: {resource_uri} ({route.method} {route.path})"
f"Registered RESOURCE: {resource_uri} ({route.method} {route.path}) with tags: {route.tags}"
)
def _create_openapi_template(self, route: openapi.HTTPRoute, operation_id: str):
@ -594,11 +605,12 @@ class FastMCPOpenAPI(FastMCP):
name=template_name,
description=enhanced_description,
parameters=template_params_schema,
tags=set(route.tags or []),
)
# Register the template by directly assigning to the templates dictionary
self._resource_manager._templates[uri_template_str] = template
logger.debug(
f"Registered TEMPLATE: {uri_template_str} ({route.method} {route.path})"
f"Registered TEMPLATE: {uri_template_str} ({route.method} {route.path}) with tags: {route.tags}"
)
async def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:

2
src/fastmcp/server/server.py
View file

@ -78,8 +78,10 @@ class FastMCP(Generic[LifespanResultT]):
lifespan: (
Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]] | None
) = None,
tags: set[str] | None = None,
**settings: Any,
):
self.tags: set[str] = tags or set()
self.settings = fastmcp.settings.ServerSettings(**settings)
self._mcp_server = MCPServer[LifespanResultT](

103
tests/server/test_openapi.py
View file

@ -36,17 +36,17 @@ def users_db() -> dict[int, User]:
def fastapi_app(users_db: dict[int, User]) -> FastAPI:
app = FastAPI(title="FastAPI App")
@app.get("/users")
@app.get("/users", tags=["users", "list"])
async def get_users() -> list[User]:
"""Get all users."""
return sorted(users_db.values(), key=lambda x: x.id)
@app.get("/users/{user_id}")
@app.get("/users/{user_id}", tags=["users", "detail"])
async def get_user(user_id: int) -> User | None:
"""Get a user by ID."""
return users_db.get(user_id)
@app.post("/users")
@app.post("/users", tags=["users", "create"])
async def create_user(user: UserCreate) -> User:
"""Create a new user."""
user_id = max(users_db.keys()) + 1
@ -54,7 +54,7 @@ def fastapi_app(users_db: dict[int, User]) -> FastAPI:
users_db[user_id] = new_user
return new_user
@app.patch("/users/{user_id}/name")
@app.patch("/users/{user_id}/name", tags=["users", "update"])
async def update_user_name(user_id: int, name: str) -> User:
"""Update a user's name."""
user = users_db.get(user_id)
@ -258,3 +258,98 @@ class TestPrompts:
"""
prompts = await fastmcp_server.list_prompts()
assert len(prompts) == 0
class TestTagTransfer:
"""Tests for transferring tags from OpenAPI to MCP objects."""
async def test_tags_transferred_to_tools(self, fastmcp_server: FastMCPOpenAPI):
"""Test that tags from OpenAPI routes are correctly transferred to Tools."""
# Get internal tools directly (not the public API which returns MCP.Content)
tools = fastmcp_server._tool_manager.list_tools()
# Find the create_user and update_user_name tools
create_user_tool = next(
(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"
),
None,
)
assert create_user_tool is not None
assert update_user_tool is not None
# Check that tags from OpenAPI routes were transferred to the Tool objects
assert "users" in create_user_tool.tags
assert "create" in create_user_tool.tags
assert len(create_user_tool.tags) == 2
assert "users" in update_user_tool.tags
assert "update" in update_user_tool.tags
assert len(update_user_tool.tags) == 2
async def test_tags_transferred_to_resources(self, fastmcp_server: FastMCPOpenAPI):
"""Test that tags from OpenAPI routes are correctly transferred to Resources."""
# Get internal resources directly
resources = fastmcp_server._resource_manager.list_resources()
# Find the get_users resource
get_users_resource = next(
(r for r in resources if r.name == "get_users_users_get"), None
)
assert get_users_resource is not None
# Check that tags from OpenAPI routes were transferred to the Resource object
assert "users" in get_users_resource.tags
assert "list" in get_users_resource.tags
assert len(get_users_resource.tags) == 2
async def test_tags_transferred_to_resource_templates(
self, fastmcp_server: FastMCPOpenAPI
):
"""Test that tags from OpenAPI routes are correctly transferred to ResourceTemplates."""
# Get internal resource templates directly
templates = fastmcp_server._resource_manager.list_templates()
# Find the get_user template
get_user_template = next(
(t for t in templates if t.name == "get_user_users__user_id__get"), None
)
assert get_user_template is not None
# Check that tags from OpenAPI routes were transferred to the ResourceTemplate object
assert "users" in get_user_template.tags
assert "detail" in get_user_template.tags
assert len(get_user_template.tags) == 2
async def test_tags_preserved_in_resources_created_from_templates(
self, fastmcp_server: FastMCPOpenAPI
):
"""Test that tags are preserved when creating resources from templates."""
# Get internal resource templates directly
templates = fastmcp_server._resource_manager.list_templates()
# Find the get_user template
get_user_template = next(
(t for t in templates if t.name == "get_user_users__user_id__get"), None
)
assert get_user_template is not None
# Manually create a resource from template
params = {"user_id": 1}
resource = await get_user_template.create_resource(
"resource://openapi/get_user_users__user_id__get/1", params
)
# Verify tags are preserved from template to resource
assert "users" in resource.tags
assert "detail" in resource.tags
assert len(resource.tags) == 2

57
tests/utilities/openapi/test_openapi.py
View file

@ -460,6 +460,63 @@ def test_petstore_required_fields_resolution(parsed_petstore_routes):
assert json_schema.get("required") == ["id", "name"]
def test_tags_parsing_in_petstore_routes(parsed_petstore_routes):
"""Test that tags are correctly parsed from the OpenAPI schema."""
# All petstore routes should have the "pets" tag
for route in parsed_petstore_routes:
assert "pets" in route.tags, (
f"Route {route.method} {route.path} is missing 'pets' tag"
)
def test_tag_list_structure(parsed_petstore_routes):
"""Test that tags are stored as a list of strings."""
for route in parsed_petstore_routes:
assert isinstance(route.tags, list), "Tags should be stored as a list"
for tag in route.tags:
assert isinstance(tag, str), "Each tag should be a string"
def test_empty_tags_handling(bookstore_schema):
"""Test that routes with no tags are handled correctly with empty lists."""
# Modify a route to remove tags
if "tags" in bookstore_schema["paths"]["/books"]["get"]:
del bookstore_schema["paths"]["/books"]["get"]["tags"]
# Parse the modified schema
routes = parse_openapi_to_http_routes(bookstore_schema)
# Find the GET /books route
get_books = next(
(r for r in routes if r.method == "GET" and r.path == "/books"), None
)
assert get_books is not None
# Should have an empty list, not None
assert get_books.tags == [], "Routes without tags should have empty tag lists"
def test_multiple_tags_preserved(bookstore_schema):
"""Test that multiple tags are preserved during parsing."""
# Add multiple tags to a route
bookstore_schema["paths"]["/books"]["get"]["tags"] = ["books", "catalog", "api"]
# Parse the modified schema
routes = parse_openapi_to_http_routes(bookstore_schema)
# Find the GET /books route
get_books = next(
(r for r in routes if r.method == "GET" and r.path == "/books"), None
)
assert get_books is not None
# Should have all tags
assert "books" in get_books.tags
assert "catalog" in get_books.tags
assert "api" in get_books.tags
assert len(get_books.tags) == 3
# --- Tests for BookStore schema --- #

88
tests/utilities/openapi/test_openapi_fastapi.py
View file

@ -432,3 +432,91 @@ def test_token_dependency_handling(route_map):
token_headers = [p for p in header_params if p.name == "x-token"]
assert len(token_headers) == 1, f"Expected x-token header in {op_id}"
assert token_headers[0].required is True
# --- Additional Tag-related Tests --- #
def test_all_routes_have_tags(parsed_routes):
"""Test that all routes have a non-empty tags list."""
for route in parsed_routes:
assert hasattr(route, "tags"), f"Route {route.path} should have tags attribute"
assert route.tags is not None, f"Route {route.path} tags should not be None"
# FastAPI adds tags to all routes in our test fixture
assert len(route.tags) > 0, f"Route {route.path} should have at least one tag"
def test_tag_consistency_across_related_endpoints(route_map):
"""Test that related endpoints have consistent tags."""
# All item endpoints should have the "items" tag
item_endpoints = [
"list_items",
"create_item",
"get_item",
"update_item",
"delete_item",
]
for endpoint in item_endpoints:
assert "items" in route_map[endpoint].tags, (
f"Endpoint {endpoint} should have 'items' tag"
)
# Tag-related endpoints should have both "items" and "tags" tags
tag_endpoints = ["update_item_tags", "get_item_tag"]
for endpoint in tag_endpoints:
assert "items" in route_map[endpoint].tags, (
f"Endpoint {endpoint} should have 'items' tag"
)
assert "tags" in route_map[endpoint].tags, (
f"Endpoint {endpoint} should have 'tags' tag"
)
def test_tag_order_preservation(fastapi_server):
"""Test that tag order is preserved in the parsed routes."""
# Add a new endpoint with specifically ordered tags
@fastapi_server.get(
"/test-tag-order",
tags=["first", "second", "third"],
operation_id="test_tag_order",
)
async def test_tag_order():
return {"result": "testing tag order"}
# Get the updated schema and parse routes
routes = parse_openapi_to_http_routes(fastapi_server.openapi())
# Find our test route
test_route = next((r for r in routes if r.path == "/test-tag-order"), None)
assert test_route is not None
# Check tag order is preserved
assert test_route.tags == ["first", "second", "third"], (
"Tag order should be preserved"
)
def test_duplicate_tags_handling(fastapi_server):
"""Test handling of duplicate tags in the OpenAPI schema."""
# Add an endpoint with duplicate tags
@fastapi_server.get(
"/test-duplicate-tags",
tags=["duplicate", "items", "duplicate"],
operation_id="test_duplicate_tags",
)
async def test_duplicate_tags():
return {"result": "testing duplicate tags"}
# Get the updated schema and parse routes
routes = parse_openapi_to_http_routes(fastapi_server.openapi())
# Find our test route
test_route = next((r for r in routes if r.path == "/test-duplicate-tags"), None)
assert test_route is not None
# Check that duplicate tags are preserved (FastAPI might deduplicate)
# We'll test both possibilities to be safe
assert "duplicate" in test_route.tags, "Tag 'duplicate' should be present"
assert "items" in test_route.tags, "Tag 'items' should be present"