From 85eff33b81248dbc7ff907822ec473f351367a2a Mon Sep 17 00:00:00 2001 From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com> Date: Fri, 6 Feb 2026 20:08:08 -0500 Subject: [PATCH] Infer MIME types from OpenAPI response definitions (#3101) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Infer mime_type from OpenAPI response content types for resources 🤖 Generated with Claude Code https://claude.ai/code/session_01FZD5ZT8WiQqfBu39ybuQis * Handle media types without schemas in MIME inference 🤖 Generated with Claude Code https://claude.ai/code/session_01FZD5ZT8WiQqfBu39ybuQis --------- Co-authored-by: Claude --- .../server/providers/openapi/components.py | 58 ++- .../server/providers/openapi/provider.py | 3 + src/fastmcp/utilities/openapi/parser.py | 4 + .../openapi/test_openapi_features.py | 357 ++++++++++++++++++ 4 files changed, 421 insertions(+), 1 deletion(-) diff --git a/src/fastmcp/server/providers/openapi/components.py b/src/fastmcp/server/providers/openapi/components.py index 43c58b956..4e6d18f1e 100644 --- a/src/fastmcp/server/providers/openapi/components.py +++ b/src/fastmcp/server/providers/openapi/components.py @@ -33,10 +33,64 @@ __all__ = [ "OpenAPIResource", "OpenAPIResourceTemplate", "OpenAPITool", + "_extract_mime_type_from_route", ] logger = get_logger(__name__) +# Default MIME type when no response content type can be inferred +_DEFAULT_MIME_TYPE = "application/json" + + +def _extract_mime_type_from_route(route: HTTPRoute) -> str: + """Extract the primary MIME type from an HTTPRoute's response definitions. + + Looks for the first successful response (2xx) and returns its content type. + Prefers JSON-compatible types when multiple are available. + Falls back to "application/json" when no response content type is declared. + """ + if not route.responses: + return _DEFAULT_MIME_TYPE + + # Priority order for success status codes + success_codes = ["200", "201", "202", "204"] + + response_info = None + for status_code in success_codes: + if status_code in route.responses: + response_info = route.responses[status_code] + break + + # If no explicit success codes, try any 2xx response + if response_info is None: + for status_code, resp_info in route.responses.items(): + if status_code.startswith("2"): + response_info = resp_info + break + + if response_info is None or not response_info.content_schema: + return _DEFAULT_MIME_TYPE + + # If there's only one content type, use it directly + content_types = list(response_info.content_schema.keys()) + if len(content_types) == 1: + return content_types[0] + + # When multiple types exist, prefer JSON-compatible types + json_compatible_types = [ + "application/json", + "application/vnd.api+json", + "application/hal+json", + "application/ld+json", + "text/json", + ] + for ct in json_compatible_types: + if ct in response_info.content_schema: + return ct + + # Fall back to the first available content type + return content_types[0] + def _slugify(text: str) -> str: """Convert text to a URL-friendly slug format. @@ -294,6 +348,7 @@ class OpenAPIResourceTemplate(ResourceTemplate): description: str, parameters: dict[str, Any], tags: set[str] | None = None, + mime_type: str = _DEFAULT_MIME_TYPE, ): super().__init__( uri_template=uri_template, @@ -301,6 +356,7 @@ class OpenAPIResourceTemplate(ResourceTemplate): description=description, parameters=parameters, tags=tags or set(), + mime_type=mime_type, ) self._client = client self._route = route @@ -325,6 +381,6 @@ class OpenAPIResourceTemplate(ResourceTemplate): uri=uri, name=f"{self.name}-{'-'.join(uri_parts)}", description=self.description or f"Resource for {self._route.path}", - mime_type="application/json", + mime_type=self.mime_type, tags=set(self._route.tags or []), ) diff --git a/src/fastmcp/server/providers/openapi/provider.py b/src/fastmcp/server/providers/openapi/provider.py index a93be1a41..ac79af400 100644 --- a/src/fastmcp/server/providers/openapi/provider.py +++ b/src/fastmcp/server/providers/openapi/provider.py @@ -17,6 +17,7 @@ from fastmcp.server.providers.openapi.components import ( OpenAPIResource, OpenAPIResourceTemplate, OpenAPITool, + _extract_mime_type_from_route, _slugify, ) from fastmcp.server.providers.openapi.routing import ( @@ -288,6 +289,7 @@ class OpenAPIProvider(Provider): uri=resource_uri, name=resource_name, description=enhanced_description, + mime_type=_extract_mime_type_from_route(route), tags=set(route.tags or []) | tags, ) @@ -356,6 +358,7 @@ class OpenAPIProvider(Provider): description=enhanced_description, parameters=template_params_schema, tags=set(route.tags or []) | tags, + mime_type=_extract_mime_type_from_route(route), ) if self._mcp_component_fn is not None: diff --git a/src/fastmcp/utilities/openapi/parser.py b/src/fastmcp/utilities/openapi/parser.py index e284295fa..40adf8d27 100644 --- a/src/fastmcp/utilities/openapi/parser.py +++ b/src/fastmcp/utilities/openapi/parser.py @@ -506,6 +506,10 @@ class OpenAPIParser( f"Failed to extract schema for media type '{media_type_str}' " f"in response {status_code}: {e}" ) + else: + # Record the media type even without a schema so MIME + # type inference can still use the declared content type. + resp_info.content_schema.setdefault(media_type_str, {}) extracted_responses[str(status_code)] = resp_info except ValueError as e: diff --git a/tests/server/providers/openapi/test_openapi_features.py b/tests/server/providers/openapi/test_openapi_features.py index b466268fe..f55a1e038 100644 --- a/tests/server/providers/openapi/test_openapi_features.py +++ b/tests/server/providers/openapi/test_openapi_features.py @@ -6,6 +6,9 @@ import pytest from fastmcp import FastMCP from fastmcp.client import Client from fastmcp.server.providers.openapi import OpenAPIProvider +from fastmcp.server.providers.openapi.components import _extract_mime_type_from_route +from fastmcp.server.providers.openapi.routing import MCPType, RouteMap +from fastmcp.utilities.openapi.models import HTTPRoute, ResponseInfo def create_openapi_server( @@ -412,3 +415,357 @@ class TestResponseSchemas: # Let's just check the tool exists and has basic properties assert get_user_tool.description is not None assert get_user_tool.name == "get_user" + + +class TestMimeTypeExtraction: + """Test MIME type extraction from route responses.""" + + def test_json_response(self): + """JSON content type is correctly extracted.""" + route = HTTPRoute( + path="/items", + method="GET", + responses={ + "200": ResponseInfo( + content_schema={"application/json": {"type": "object"}} + ) + }, + ) + assert _extract_mime_type_from_route(route) == "application/json" + + def test_text_plain_response(self): + """Plain text content type is correctly extracted.""" + route = HTTPRoute( + path="/health", + method="GET", + responses={ + "200": ResponseInfo(content_schema={"text/plain": {"type": "string"}}) + }, + ) + assert _extract_mime_type_from_route(route) == "text/plain" + + def test_text_html_response(self): + """HTML content type is correctly extracted.""" + route = HTTPRoute( + path="/page", + method="GET", + responses={ + "200": ResponseInfo(content_schema={"text/html": {"type": "string"}}) + }, + ) + assert _extract_mime_type_from_route(route) == "text/html" + + def test_image_response(self): + """Image content type is correctly extracted.""" + route = HTTPRoute( + path="/avatar", + method="GET", + responses={ + "200": ResponseInfo( + content_schema={"image/png": {"type": "string", "format": "binary"}} + ) + }, + ) + assert _extract_mime_type_from_route(route) == "image/png" + + def test_no_responses_defaults_to_json(self): + """Empty responses default to application/json.""" + route = HTTPRoute(path="/items", method="GET", responses={}) + assert _extract_mime_type_from_route(route) == "application/json" + + def test_no_content_schema_defaults_to_json(self): + """Response without content_schema defaults to application/json.""" + route = HTTPRoute( + path="/items", + method="GET", + responses={"204": ResponseInfo(description="No content")}, + ) + assert _extract_mime_type_from_route(route) == "application/json" + + def test_prefers_json_when_multiple_types(self): + """When both JSON and other types exist, JSON is preferred.""" + route = HTTPRoute( + path="/items", + method="GET", + responses={ + "200": ResponseInfo( + content_schema={ + "text/html": {"type": "string"}, + "application/json": {"type": "object"}, + } + ) + }, + ) + assert _extract_mime_type_from_route(route) == "application/json" + + def test_non_standard_2xx_code(self): + """Falls back to any 2xx status code when standard ones are missing.""" + route = HTTPRoute( + path="/items", + method="GET", + responses={ + "206": ResponseInfo( + content_schema={ + "application/octet-stream": { + "type": "string", + "format": "binary", + } + } + ) + }, + ) + assert _extract_mime_type_from_route(route) == "application/octet-stream" + + def test_ignores_error_responses(self): + """Only error responses (no 2xx) results in default.""" + route = HTTPRoute( + path="/items", + method="GET", + responses={ + "404": ResponseInfo( + content_schema={"application/json": {"type": "object"}} + ) + }, + ) + assert _extract_mime_type_from_route(route) == "application/json" + + def test_201_response(self): + """201 Created response content type is extracted.""" + route = HTTPRoute( + path="/items", + method="POST", + responses={ + "201": ResponseInfo(content_schema={"text/plain": {"type": "string"}}) + }, + ) + assert _extract_mime_type_from_route(route) == "text/plain" + + def test_media_type_without_schema(self): + """Media type declared without a schema still infers MIME type.""" + route = HTTPRoute( + path="/health", + method="GET", + responses={"200": ResponseInfo(content_schema={"text/plain": {}})}, + ) + assert _extract_mime_type_from_route(route) == "text/plain" + + +class TestResourceTemplateMimeType: + """Test that OpenAPIResourceTemplate uses inferred MIME types.""" + + @pytest.fixture + def text_plain_spec(self): + """OpenAPI spec with a text/plain resource template endpoint.""" + return { + "openapi": "3.0.0", + "info": {"title": "Text API", "version": "1.0.0"}, + "servers": [{"url": "https://api.example.com"}], + "paths": { + "/documents/{id}": { + "get": { + "operationId": "get_document", + "summary": "Get document content", + "parameters": [ + { + "name": "id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ], + "responses": { + "200": { + "description": "Document content", + "content": { + "text/plain": {"schema": {"type": "string"}} + }, + } + }, + } + } + }, + } + + @pytest.fixture + def html_spec(self): + """OpenAPI spec with a text/html resource endpoint.""" + return { + "openapi": "3.0.0", + "info": {"title": "HTML API", "version": "1.0.0"}, + "servers": [{"url": "https://api.example.com"}], + "paths": { + "/pages/{slug}": { + "get": { + "operationId": "get_page", + "summary": "Get HTML page", + "parameters": [ + { + "name": "slug", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ], + "responses": { + "200": { + "description": "HTML page", + "content": { + "text/html": {"schema": {"type": "string"}} + }, + } + }, + } + } + }, + } + + async def test_resource_template_text_plain_mime_type(self, text_plain_spec): + """Resource template should reflect text/plain from OpenAPI spec.""" + route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] + async with httpx.AsyncClient(base_url="https://api.example.com") as client: + provider = OpenAPIProvider( + openapi_spec=text_plain_spec, client=client, route_maps=route_maps + ) + mcp = FastMCP("Test") + mcp.add_provider(provider) + async with Client(mcp) as mcp_client: + templates = await mcp_client.list_resource_templates() + assert len(templates) == 1 + assert templates[0].mimeType == "text/plain" + + async def test_resource_template_html_mime_type(self, html_spec): + """Resource template should reflect text/html from OpenAPI spec.""" + route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] + async with httpx.AsyncClient(base_url="https://api.example.com") as client: + provider = OpenAPIProvider( + openapi_spec=html_spec, client=client, route_maps=route_maps + ) + mcp = FastMCP("Test") + mcp.add_provider(provider) + async with Client(mcp) as mcp_client: + templates = await mcp_client.list_resource_templates() + assert len(templates) == 1 + assert templates[0].mimeType == "text/html" + + async def test_resource_template_defaults_json_mime_type(self): + """Resource template defaults to application/json for JSON responses.""" + spec = { + "openapi": "3.0.0", + "info": {"title": "JSON API", "version": "1.0.0"}, + "servers": [{"url": "https://api.example.com"}], + "paths": { + "/users/{id}": { + "get": { + "operationId": "get_user", + "summary": "Get user", + "parameters": [ + { + "name": "id", + "in": "path", + "required": True, + "schema": {"type": "integer"}, + } + ], + "responses": { + "200": { + "description": "User data", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "name": {"type": "string"}, + }, + } + } + }, + } + }, + } + } + }, + } + route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] + async with httpx.AsyncClient(base_url="https://api.example.com") as client: + provider = OpenAPIProvider( + openapi_spec=spec, client=client, route_maps=route_maps + ) + mcp = FastMCP("Test") + mcp.add_provider(provider) + async with Client(mcp) as mcp_client: + templates = await mcp_client.list_resource_templates() + assert len(templates) == 1 + assert templates[0].mimeType == "application/json" + + +class TestResourceMimeType: + """Test that OpenAPIResource uses inferred MIME types.""" + + async def test_resource_text_plain_mime_type(self): + """Static resource should reflect text/plain from OpenAPI spec.""" + spec = { + "openapi": "3.0.0", + "info": {"title": "Health API", "version": "1.0.0"}, + "servers": [{"url": "https://api.example.com"}], + "paths": { + "/health": { + "get": { + "operationId": "healthcheck", + "summary": "Health check", + "responses": { + "200": { + "description": "Health status", + "content": { + "text/plain": {"schema": {"type": "string"}} + }, + } + }, + } + } + }, + } + route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE)] + async with httpx.AsyncClient(base_url="https://api.example.com") as client: + provider = OpenAPIProvider( + openapi_spec=spec, client=client, route_maps=route_maps + ) + mcp = FastMCP("Test") + mcp.add_provider(provider) + async with Client(mcp) as mcp_client: + resources = await mcp_client.list_resources() + assert len(resources) == 1 + assert resources[0].mimeType == "text/plain" + + async def test_resource_mime_type_without_schema(self): + """Resource with media type but no schema still infers MIME type.""" + spec = { + "openapi": "3.0.0", + "info": {"title": "Health API", "version": "1.0.0"}, + "servers": [{"url": "https://api.example.com"}], + "paths": { + "/health": { + "get": { + "operationId": "healthcheck", + "summary": "Health check", + "responses": { + "200": { + "description": "Health status", + "content": {"text/plain": {}}, + } + }, + } + } + }, + } + route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE)] + async with httpx.AsyncClient(base_url="https://api.example.com") as client: + provider = OpenAPIProvider( + openapi_spec=spec, client=client, route_maps=route_maps + ) + mcp = FastMCP("Test") + mcp.add_provider(provider) + async with Client(mcp) as mcp_client: + resources = await mcp_client.list_resources() + assert len(resources) == 1 + assert resources[0].mimeType == "text/plain"