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

Infer MIME types from OpenAPI response definitions (#3101)

Browse source 
* 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 <noreply@anthropic.com>
This commit is contained in:
Jeremiah Lowin 2026-02-06 20:08:08 -05:00 • committed by GitHub
commit 85eff33b81
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 421 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

58
src/fastmcp/server/providers/openapi/components.py
View file

@ -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 []),
)

3
src/fastmcp/server/providers/openapi/provider.py
View file

@ -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:

4
src/fastmcp/utilities/openapi/parser.py
View file

@ -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:

357
tests/server/providers/openapi/test_openapi_features.py
View file

@ -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"