mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-23 22:14:18 +02:00
Merge pull request #293 from jlowin/openapi-descriptions
Ensure openapi descriptions are included in tool details
This commit is contained in:
Jeremiah Lowin
GitHub
commit
9d422e08aa
3 changed files with 904 additions and 40 deletions
|
|
@ -534,10 +534,12 @@ class FastMCPOpenAPI(FastMCP):
|
|||
or f"Executes {route.method} {route.path}"
|
||||
)
|
||||
|
||||
# Format enhanced description
|
||||
# Format enhanced description with parameters and request body
|
||||
enhanced_description = format_description_with_responses(
|
||||
base_description=base_description,
|
||||
responses=route.responses,
|
||||
parameters=route.parameters,
|
||||
request_body=route.request_body,
|
||||
)
|
||||
|
||||
tool = OpenAPITool(
|
||||
|
|
@ -565,10 +567,12 @@ class FastMCPOpenAPI(FastMCP):
|
|||
route.description or route.summary or f"Represents {route.path}"
|
||||
)
|
||||
|
||||
# Format enhanced description
|
||||
# Format enhanced description with parameters and request body
|
||||
enhanced_description = format_description_with_responses(
|
||||
base_description=base_description,
|
||||
responses=route.responses,
|
||||
parameters=route.parameters,
|
||||
request_body=route.request_body,
|
||||
)
|
||||
|
||||
resource = OpenAPIResource(
|
||||
|
|
@ -600,16 +604,30 @@ class FastMCPOpenAPI(FastMCP):
|
|||
route.description or route.summary or f"Template for {route.path}"
|
||||
)
|
||||
|
||||
# Format enhanced description
|
||||
# Format enhanced description with parameters and request body
|
||||
enhanced_description = format_description_with_responses(
|
||||
base_description=base_description,
|
||||
responses=route.responses,
|
||||
parameters=route.parameters,
|
||||
request_body=route.request_body,
|
||||
)
|
||||
|
||||
template_params_schema = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
p.name: p.schema_ for p in route.parameters if p.location == "path"
|
||||
p.name: {
|
||||
**(p.schema_.copy() if isinstance(p.schema_, dict) else {}),
|
||||
**(
|
||||
{"description": p.description}
|
||||
if p.description
|
||||
and not (
|
||||
isinstance(p.schema_, dict) and "description" in p.schema_
|
||||
)
|
||||
else {}
|
||||
),
|
||||
}
|
||||
for p in route.parameters
|
||||
if p.location == "path"
|
||||
},
|
||||
"required": [
|
||||
p.name for p in route.parameters if p.location == "path" and p.required
|
||||
|
|
|
|||
|
|
@ -1001,53 +1001,153 @@ def format_description_with_responses(
|
|||
responses: dict[
|
||||
str, Any
|
||||
], # Changed from specific ResponseInfo type to avoid circular imports
|
||||
parameters: list[openapi.ParameterInfo] | None = None, # Add parameters parameter
|
||||
request_body: openapi.RequestBodyInfo | None = None, # Add request_body parameter
|
||||
) -> str:
|
||||
"""Formats the base description string with response information."""
|
||||
if not responses:
|
||||
return base_description
|
||||
"""
|
||||
Formats the base description string with response, parameter, and request body information.
|
||||
|
||||
Args:
|
||||
base_description (str): The initial description to be formatted.
|
||||
responses (dict[str, Any]): A dictionary of response information, keyed by status code.
|
||||
parameters (list[openapi.ParameterInfo] | None, optional): A list of parameter information,
|
||||
including path and query parameters. Each parameter includes details such as name,
|
||||
location, whether it is required, and a description.
|
||||
request_body (openapi.RequestBodyInfo | None, optional): Information about the request body,
|
||||
including its description, whether it is required, and its content schema.
|
||||
|
||||
Returns:
|
||||
str: The formatted description string with additional details about responses, parameters,
|
||||
and the request body.
|
||||
"""
|
||||
desc_parts = [base_description]
|
||||
response_section = "\n\n**Responses:**"
|
||||
added_response_section = False
|
||||
|
||||
# Determine success codes (common ones)
|
||||
success_codes = {"200", "201", "202", "204"} # As strings
|
||||
success_status = next((s for s in success_codes if s in responses), None)
|
||||
# Add parameter information
|
||||
if parameters:
|
||||
# Process path parameters
|
||||
path_params = [p for p in parameters if p.location == "path"]
|
||||
if path_params:
|
||||
param_section = "\n\n**Path Parameters:**"
|
||||
desc_parts.append(param_section)
|
||||
for param in path_params:
|
||||
required_marker = " (Required)" if param.required else ""
|
||||
param_desc = f"\n- **{param.name}**{required_marker}: {param.description or 'No description.'}"
|
||||
desc_parts.append(param_desc)
|
||||
|
||||
# Process all responses
|
||||
responses_to_process = responses.items()
|
||||
# Process query parameters
|
||||
query_params = [p for p in parameters if p.location == "query"]
|
||||
if query_params:
|
||||
param_section = "\n\n**Query Parameters:**"
|
||||
desc_parts.append(param_section)
|
||||
for param in query_params:
|
||||
required_marker = " (Required)" if param.required else ""
|
||||
param_desc = f"\n- **{param.name}**{required_marker}: {param.description or 'No description.'}"
|
||||
desc_parts.append(param_desc)
|
||||
|
||||
for status_code, resp_info in sorted(responses_to_process):
|
||||
if not added_response_section:
|
||||
desc_parts.append(response_section)
|
||||
added_response_section = True
|
||||
# Add request body information if present
|
||||
if request_body and request_body.description:
|
||||
req_body_section = "\n\n**Request Body:**"
|
||||
desc_parts.append(req_body_section)
|
||||
required_marker = " (Required)" if request_body.required else ""
|
||||
desc_parts.append(f"\n{request_body.description}{required_marker}")
|
||||
|
||||
status_marker = " (Success)" if status_code == success_status else ""
|
||||
desc_parts.append(
|
||||
f"\n- **{status_code}**{status_marker}: {resp_info.description or 'No description.'}"
|
||||
)
|
||||
|
||||
# Process content schemas for this response
|
||||
if resp_info.content_schema:
|
||||
# Prioritize json, then take first available
|
||||
# Add request body property descriptions if available
|
||||
if request_body.content_schema:
|
||||
media_type = (
|
||||
"application/json"
|
||||
if "application/json" in resp_info.content_schema
|
||||
else next(iter(resp_info.content_schema), None)
|
||||
if "application/json" in request_body.content_schema
|
||||
else next(iter(request_body.content_schema), None)
|
||||
)
|
||||
if media_type:
|
||||
schema = request_body.content_schema.get(media_type, {})
|
||||
if isinstance(schema, dict) and "properties" in schema:
|
||||
desc_parts.append("\n\n**Request Properties:**")
|
||||
for prop_name, prop_schema in schema["properties"].items():
|
||||
if (
|
||||
isinstance(prop_schema, dict)
|
||||
and "description" in prop_schema
|
||||
):
|
||||
required = prop_name in schema.get("required", [])
|
||||
req_mark = " (Required)" if required else ""
|
||||
desc_parts.append(
|
||||
f"\n- **{prop_name}**{req_mark}: {prop_schema['description']}"
|
||||
)
|
||||
|
||||
# Add response information
|
||||
if responses:
|
||||
response_section = "\n\n**Responses:**"
|
||||
added_response_section = False
|
||||
|
||||
# Determine success codes (common ones)
|
||||
success_codes = {"200", "201", "202", "204"} # As strings
|
||||
success_status = next((s for s in success_codes if s in responses), None)
|
||||
|
||||
# Process all responses
|
||||
responses_to_process = responses.items()
|
||||
|
||||
for status_code, resp_info in sorted(responses_to_process):
|
||||
if not added_response_section:
|
||||
desc_parts.append(response_section)
|
||||
added_response_section = True
|
||||
|
||||
status_marker = " (Success)" if status_code == success_status else ""
|
||||
desc_parts.append(
|
||||
f"\n- **{status_code}**{status_marker}: {resp_info.description or 'No description.'}"
|
||||
)
|
||||
|
||||
if media_type:
|
||||
schema = resp_info.content_schema.get(media_type)
|
||||
desc_parts.append(f" - Content-Type: `{media_type}`")
|
||||
# Process content schemas for this response
|
||||
if resp_info.content_schema:
|
||||
# Prioritize json, then take first available
|
||||
media_type = (
|
||||
"application/json"
|
||||
if "application/json" in resp_info.content_schema
|
||||
else next(iter(resp_info.content_schema), None)
|
||||
)
|
||||
|
||||
if media_type:
|
||||
schema = resp_info.content_schema.get(media_type)
|
||||
desc_parts.append(f" - Content-Type: `{media_type}`")
|
||||
|
||||
# Add response property descriptions
|
||||
if isinstance(schema, dict):
|
||||
# Handle array responses
|
||||
if schema.get("type") == "array" and "items" in schema:
|
||||
items_schema = schema["items"]
|
||||
if (
|
||||
isinstance(items_schema, dict)
|
||||
and "properties" in items_schema
|
||||
):
|
||||
desc_parts.append("\n - **Response Item Properties:**")
|
||||
for prop_name, prop_schema in items_schema[
|
||||
"properties"
|
||||
].items():
|
||||
if (
|
||||
isinstance(prop_schema, dict)
|
||||
and "description" in prop_schema
|
||||
):
|
||||
desc_parts.append(
|
||||
f"\n - **{prop_name}**: {prop_schema['description']}"
|
||||
)
|
||||
# Handle object responses
|
||||
elif "properties" in schema:
|
||||
desc_parts.append("\n - **Response Properties:**")
|
||||
for prop_name, prop_schema in schema["properties"].items():
|
||||
if (
|
||||
isinstance(prop_schema, dict)
|
||||
and "description" in prop_schema
|
||||
):
|
||||
desc_parts.append(
|
||||
f"\n - **{prop_name}**: {prop_schema['description']}"
|
||||
)
|
||||
|
||||
if schema:
|
||||
# Generate Example
|
||||
example = generate_example_from_schema(schema)
|
||||
if example != "unknown_type" and example is not None:
|
||||
desc_parts.append("\n - **Example:**")
|
||||
desc_parts.append(
|
||||
format_json_for_description(example, indent=2)
|
||||
)
|
||||
if schema:
|
||||
example = generate_example_from_schema(schema)
|
||||
if example != "unknown_type" and example is not None:
|
||||
desc_parts.append("\n - **Example:**")
|
||||
desc_parts.append(
|
||||
format_json_for_description(example, indent=2)
|
||||
)
|
||||
|
||||
return "\n".join(desc_parts)
|
||||
|
||||
|
|
@ -1069,7 +1169,15 @@ def _combine_schemas(route: openapi.HTTPRoute) -> dict[str, Any]:
|
|||
for param in route.parameters:
|
||||
if param.required:
|
||||
required.append(param.name)
|
||||
properties[param.name] = param.schema_
|
||||
|
||||
# Copy the schema and add description if available
|
||||
param_schema = param.schema_.copy() if isinstance(param.schema_, dict) else {}
|
||||
|
||||
# Add parameter description to schema if available and not already present
|
||||
if param.description and not param_schema.get("description"):
|
||||
param_schema["description"] = param.description
|
||||
|
||||
properties[param.name] = param_schema
|
||||
|
||||
# Add request body if it exists
|
||||
if route.request_body and route.request_body.content_schema:
|
||||
|
|
@ -1077,8 +1185,11 @@ def _combine_schemas(route: openapi.HTTPRoute) -> dict[str, Any]:
|
|||
content_type = next(iter(route.request_body.content_schema))
|
||||
body_schema = route.request_body.content_schema[content_type]
|
||||
body_props = body_schema.get("properties", {})
|
||||
|
||||
# Add request body properties
|
||||
for prop_name, prop_schema in body_props.items():
|
||||
properties[prop_name] = prop_schema
|
||||
|
||||
if route.request_body.required:
|
||||
required.extend(body_schema.get("required", []))
|
||||
|
||||
|
|
|
|||
|
|
@ -1034,3 +1034,738 @@ async def test_none_path_parameters_rejected(
|
|||
"name": "New Name",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class TestDescriptionPropagation:
|
||||
"""Tests for OpenAPI description propagation to FastMCP components.
|
||||
|
||||
Each test focuses on a single, specific behavior to make it immediately clear
|
||||
what's broken when a test fails.
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def simple_openapi_spec(self) -> dict:
|
||||
"""Create a minimal OpenAPI spec with obvious test descriptions."""
|
||||
return {
|
||||
"openapi": "3.1.0",
|
||||
"info": {"title": "Test API", "version": "1.0.0"},
|
||||
"paths": {
|
||||
"/items": {
|
||||
"get": {
|
||||
"operationId": "listItems",
|
||||
"summary": "List items summary",
|
||||
"description": "LIST_DESCRIPTION\n\nFUNCTION_LIST_DESCRIPTION",
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "LIST_RESPONSE_DESCRIPTION",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {
|
||||
"type": "string",
|
||||
"description": "ITEM_RESPONSE_ID_DESCRIPTION",
|
||||
},
|
||||
"name": {
|
||||
"type": "string",
|
||||
"description": "ITEM_RESPONSE_NAME_DESCRIPTION",
|
||||
},
|
||||
"price": {
|
||||
"type": "number",
|
||||
"description": "ITEM_RESPONSE_PRICE_DESCRIPTION",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
"/items/{item_id}": {
|
||||
"get": {
|
||||
"operationId": "getItem",
|
||||
"summary": "Get item summary",
|
||||
"description": "GET_DESCRIPTION\n\nFUNCTION_GET_DESCRIPTION",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "item_id",
|
||||
"in": "path",
|
||||
"required": True,
|
||||
"description": "PATH_PARAM_DESCRIPTION",
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
{
|
||||
"name": "fields",
|
||||
"in": "query",
|
||||
"required": False,
|
||||
"description": "QUERY_PARAM_DESCRIPTION",
|
||||
"schema": {"type": "string"},
|
||||
},
|
||||
],
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "GET_RESPONSE_DESCRIPTION",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {
|
||||
"type": "string",
|
||||
"description": "ITEM_RESPONSE_ID_DESCRIPTION",
|
||||
},
|
||||
"name": {
|
||||
"type": "string",
|
||||
"description": "ITEM_RESPONSE_NAME_DESCRIPTION",
|
||||
},
|
||||
"price": {
|
||||
"type": "number",
|
||||
"description": "ITEM_RESPONSE_PRICE_DESCRIPTION",
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
"/items/create": {
|
||||
"post": {
|
||||
"operationId": "createItem",
|
||||
"summary": "Create item summary",
|
||||
"description": "CREATE_DESCRIPTION\n\nFUNCTION_CREATE_DESCRIPTION",
|
||||
"requestBody": {
|
||||
"required": True,
|
||||
"description": "BODY_DESCRIPTION",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {
|
||||
"type": "string",
|
||||
"description": "PROP_DESCRIPTION",
|
||||
}
|
||||
},
|
||||
"required": ["name"],
|
||||
}
|
||||
}
|
||||
},
|
||||
},
|
||||
"responses": {
|
||||
"201": {
|
||||
"description": "CREATE_RESPONSE_DESCRIPTION",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"id": {
|
||||
"type": "string",
|
||||
"description": "ITEM_RESPONSE_ID_DESCRIPTION",
|
||||
},
|
||||
"name": {
|
||||
"type": "string",
|
||||
"description": "ITEM_RESPONSE_NAME_DESCRIPTION",
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
async def mock_client(self) -> httpx.AsyncClient:
|
||||
"""Create a mock client that returns simple responses."""
|
||||
|
||||
async def _responder(request):
|
||||
if request.url.path == "/items" and request.method == "GET":
|
||||
return httpx.Response(200, json=[{"id": "1", "name": "Item 1"}])
|
||||
elif request.url.path.startswith("/items/") and request.method == "GET":
|
||||
item_id = request.url.path.split("/")[-1]
|
||||
return httpx.Response(
|
||||
200, json={"id": item_id, "name": f"Item {item_id}"}
|
||||
)
|
||||
elif request.url.path == "/items/create" and request.method == "POST":
|
||||
import json
|
||||
|
||||
data = json.loads(request.content)
|
||||
return httpx.Response(201, json={"id": "new", "name": data.get("name")})
|
||||
|
||||
return httpx.Response(404)
|
||||
|
||||
transport = httpx.MockTransport(_responder)
|
||||
return httpx.AsyncClient(transport=transport, base_url="http://test")
|
||||
|
||||
@pytest.fixture
|
||||
async def test_server(self, simple_openapi_spec, mock_client):
|
||||
"""Create a FastMCPOpenAPI server with the simple test spec."""
|
||||
return FastMCPOpenAPI(
|
||||
openapi_spec=simple_openapi_spec,
|
||||
client=mock_client,
|
||||
name="Test API",
|
||||
)
|
||||
|
||||
# --- RESOURCE TESTS ---
|
||||
|
||||
async def test_resource_includes_route_description(self, test_server):
|
||||
"""Test that a Resource includes the route description."""
|
||||
resources = list(test_server._resource_manager.get_resources().values())
|
||||
list_resource = next((r for r in resources if r.name == "listItems"), None)
|
||||
|
||||
assert list_resource is not None, "listItems resource wasn't created"
|
||||
assert "LIST_DESCRIPTION" in (list_resource.description or ""), (
|
||||
"Route description missing from Resource"
|
||||
)
|
||||
|
||||
async def test_resource_includes_response_description(self, test_server):
|
||||
"""Test that a Resource includes the response description."""
|
||||
resources = list(test_server._resource_manager.get_resources().values())
|
||||
list_resource = next((r for r in resources if r.name == "listItems"), None)
|
||||
|
||||
assert list_resource is not None, "listItems resource wasn't created"
|
||||
assert "LIST_RESPONSE_DESCRIPTION" in (list_resource.description or ""), (
|
||||
"Response description missing from Resource"
|
||||
)
|
||||
|
||||
async def test_resource_includes_response_model_fields(self, test_server):
|
||||
"""Test that a Resource description includes response model field descriptions."""
|
||||
resources = list(test_server._resource_manager.get_resources().values())
|
||||
list_resource = next((r for r in resources if r.name == "listItems"), None)
|
||||
|
||||
assert list_resource is not None, "listItems resource wasn't created"
|
||||
description = list_resource.description or ""
|
||||
assert "ITEM_RESPONSE_ID_DESCRIPTION" in description, (
|
||||
"Response model field descriptions missing from Resource description"
|
||||
)
|
||||
assert "ITEM_RESPONSE_NAME_DESCRIPTION" in description, (
|
||||
"Response model field descriptions missing from Resource description"
|
||||
)
|
||||
assert "ITEM_RESPONSE_PRICE_DESCRIPTION" in description, (
|
||||
"Response model field descriptions missing from Resource description"
|
||||
)
|
||||
|
||||
# --- RESOURCE TEMPLATE TESTS ---
|
||||
|
||||
async def test_template_includes_route_description(self, test_server):
|
||||
"""Test that a ResourceTemplate includes the route description."""
|
||||
templates = list(test_server._resource_manager.get_templates().values())
|
||||
get_template = next((t for t in templates if t.name == "getItem"), None)
|
||||
|
||||
assert get_template is not None, "getItem template wasn't created"
|
||||
assert "GET_DESCRIPTION" in (get_template.description or ""), (
|
||||
"Route description missing from ResourceTemplate"
|
||||
)
|
||||
|
||||
async def test_template_includes_function_docstring(self, test_server):
|
||||
"""Test that a ResourceTemplate includes the function docstring."""
|
||||
templates = list(test_server._resource_manager.get_templates().values())
|
||||
get_template = next((t for t in templates if t.name == "getItem"), None)
|
||||
|
||||
assert get_template is not None, "getItem template wasn't created"
|
||||
assert "FUNCTION_GET_DESCRIPTION" in (get_template.description or ""), (
|
||||
"Function docstring missing from ResourceTemplate"
|
||||
)
|
||||
|
||||
async def test_template_includes_path_parameter_description(self, test_server):
|
||||
"""Test that a ResourceTemplate includes path parameter descriptions."""
|
||||
templates = list(test_server._resource_manager.get_templates().values())
|
||||
get_template = next((t for t in templates if t.name == "getItem"), None)
|
||||
|
||||
assert get_template is not None, "getItem template wasn't created"
|
||||
assert "PATH_PARAM_DESCRIPTION" in (get_template.description or ""), (
|
||||
"Path parameter description missing from ResourceTemplate description"
|
||||
)
|
||||
|
||||
async def test_template_includes_query_parameter_description(self, test_server):
|
||||
"""Test that a ResourceTemplate includes query parameter descriptions."""
|
||||
templates = list(test_server._resource_manager.get_templates().values())
|
||||
get_template = next((t for t in templates if t.name == "getItem"), None)
|
||||
|
||||
assert get_template is not None, "getItem template wasn't created"
|
||||
assert "QUERY_PARAM_DESCRIPTION" in (get_template.description or ""), (
|
||||
"Query parameter description missing from ResourceTemplate description"
|
||||
)
|
||||
|
||||
async def test_template_parameter_schema_includes_description(self, test_server):
|
||||
"""Test that a ResourceTemplate's parameter schema includes parameter descriptions."""
|
||||
templates = list(test_server._resource_manager.get_templates().values())
|
||||
get_template = next((t for t in templates if t.name == "getItem"), None)
|
||||
|
||||
assert get_template is not None, "getItem template wasn't created"
|
||||
assert "properties" in get_template.parameters, (
|
||||
"Schema properties missing from ResourceTemplate"
|
||||
)
|
||||
assert "item_id" in get_template.parameters["properties"], (
|
||||
"item_id missing from ResourceTemplate schema"
|
||||
)
|
||||
assert "description" in get_template.parameters["properties"]["item_id"], (
|
||||
"Description missing from item_id parameter schema"
|
||||
)
|
||||
assert (
|
||||
"PATH_PARAM_DESCRIPTION"
|
||||
in get_template.parameters["properties"]["item_id"]["description"]
|
||||
), "Path parameter description incorrect in schema"
|
||||
|
||||
# --- TOOL TESTS ---
|
||||
|
||||
async def test_tool_includes_route_description(self, test_server):
|
||||
"""Test that a Tool includes the route description."""
|
||||
tools = test_server._tool_manager.list_tools()
|
||||
create_tool = next((t for t in tools if t.name == "createItem"), None)
|
||||
|
||||
assert create_tool is not None, "createItem tool wasn't created"
|
||||
assert "CREATE_DESCRIPTION" in (create_tool.description or ""), (
|
||||
"Route description missing from Tool"
|
||||
)
|
||||
|
||||
async def test_tool_includes_function_docstring(self, test_server):
|
||||
"""Test that a Tool includes the function docstring."""
|
||||
tools = test_server._tool_manager.list_tools()
|
||||
create_tool = next((t for t in tools if t.name == "createItem"), None)
|
||||
|
||||
assert create_tool is not None, "createItem tool wasn't created"
|
||||
description = create_tool.description or ""
|
||||
assert "FUNCTION_CREATE_DESCRIPTION" in description, (
|
||||
"Function docstring missing from Tool"
|
||||
)
|
||||
|
||||
async def test_tool_parameter_schema_includes_property_description(
|
||||
self, test_server
|
||||
):
|
||||
"""Test that a Tool's parameter schema includes property descriptions from request model."""
|
||||
tools = test_server._tool_manager.list_tools()
|
||||
create_tool = next((t for t in tools if t.name == "createItem"), None)
|
||||
|
||||
assert create_tool is not None, "createItem tool wasn't created"
|
||||
assert "properties" in create_tool.parameters, (
|
||||
"Schema properties missing from Tool"
|
||||
)
|
||||
assert "name" in create_tool.parameters["properties"], (
|
||||
"name parameter missing from Tool schema"
|
||||
)
|
||||
assert "description" in create_tool.parameters["properties"]["name"], (
|
||||
"Description missing from name parameter schema"
|
||||
)
|
||||
assert (
|
||||
"PROP_DESCRIPTION"
|
||||
in create_tool.parameters["properties"]["name"]["description"]
|
||||
), "Property description incorrect in schema"
|
||||
|
||||
# --- CLIENT API TESTS ---
|
||||
|
||||
async def test_client_api_resource_description(self, test_server):
|
||||
"""Test that Resource descriptions are accessible via the client API."""
|
||||
async with Client(test_server) as client:
|
||||
resources = await client.list_resources()
|
||||
list_resource = next((r for r in resources if r.name == "listItems"), None)
|
||||
|
||||
assert list_resource is not None, (
|
||||
"listItems resource not accessible via client API"
|
||||
)
|
||||
resource_description = list_resource.description or ""
|
||||
assert "LIST_DESCRIPTION" in resource_description, (
|
||||
"Route description missing in Resource from client API"
|
||||
)
|
||||
|
||||
async def test_client_api_template_description(self, test_server):
|
||||
"""Test that ResourceTemplate descriptions are accessible via the client API."""
|
||||
async with Client(test_server) as client:
|
||||
templates = await client.list_resource_templates()
|
||||
get_template = next((t for t in templates if t.name == "getItem"), None)
|
||||
|
||||
assert get_template is not None, (
|
||||
"getItem template not accessible via client API"
|
||||
)
|
||||
template_description = get_template.description or ""
|
||||
assert "GET_DESCRIPTION" in template_description, (
|
||||
"Route description missing in ResourceTemplate from client API"
|
||||
)
|
||||
|
||||
async def test_client_api_tool_description(self, test_server):
|
||||
"""Test that Tool descriptions are accessible via the client API."""
|
||||
async with Client(test_server) as client:
|
||||
tools = await client.list_tools()
|
||||
create_tool = next((t for t in tools if t.name == "createItem"), None)
|
||||
|
||||
assert create_tool is not None, (
|
||||
"createItem tool not accessible via client API"
|
||||
)
|
||||
tool_description = create_tool.description or ""
|
||||
assert "FUNCTION_CREATE_DESCRIPTION" in tool_description, (
|
||||
"Function docstring missing in Tool from client API"
|
||||
)
|
||||
|
||||
async def test_client_api_tool_parameter_schema(self, test_server):
|
||||
"""Test that Tool parameter schemas are accessible via the client API."""
|
||||
async with Client(test_server) as client:
|
||||
tools = await client.list_tools()
|
||||
create_tool = next((t for t in tools if t.name == "createItem"), None)
|
||||
|
||||
assert create_tool is not None, (
|
||||
"createItem tool not accessible via client API"
|
||||
)
|
||||
assert "properties" in create_tool.inputSchema, (
|
||||
"Schema properties missing from Tool inputSchema in client API"
|
||||
)
|
||||
assert "name" in create_tool.inputSchema["properties"], (
|
||||
"name parameter missing from Tool schema in client API"
|
||||
)
|
||||
assert "description" in create_tool.inputSchema["properties"]["name"], (
|
||||
"Description missing from name parameter in client API"
|
||||
)
|
||||
assert (
|
||||
"PROP_DESCRIPTION"
|
||||
in create_tool.inputSchema["properties"]["name"]["description"]
|
||||
), "Property description incorrect in schema from client API"
|
||||
|
||||
|
||||
class TestFastAPIDescriptionPropagation:
|
||||
"""Tests for FastAPI docstring and annotation propagation to FastMCP components.
|
||||
|
||||
Each test focuses on a single, specific behavior to make it immediately clear
|
||||
what's broken when a test fails.
|
||||
"""
|
||||
|
||||
@pytest.fixture
|
||||
def fastapi_app_with_descriptions(self) -> FastAPI:
|
||||
"""Create a simple FastAPI app with docstrings and annotations."""
|
||||
from typing import Annotated
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
app = FastAPI(title="Test FastAPI App")
|
||||
|
||||
class Item(BaseModel):
|
||||
name: str = Field(..., description="ITEM_NAME_DESCRIPTION")
|
||||
price: float = Field(..., description="ITEM_PRICE_DESCRIPTION")
|
||||
|
||||
class ItemResponse(BaseModel):
|
||||
id: str = Field(..., description="ITEM_RESPONSE_ID_DESCRIPTION")
|
||||
name: str = Field(..., description="ITEM_RESPONSE_NAME_DESCRIPTION")
|
||||
price: float = Field(..., description="ITEM_RESPONSE_PRICE_DESCRIPTION")
|
||||
|
||||
@app.get("/items", tags=["items"])
|
||||
async def list_items() -> list[ItemResponse]:
|
||||
"""FUNCTION_LIST_DESCRIPTION
|
||||
|
||||
Returns a list of items.
|
||||
"""
|
||||
return [
|
||||
ItemResponse(id="1", name="Item 1", price=10.0),
|
||||
ItemResponse(id="2", name="Item 2", price=20.0),
|
||||
]
|
||||
|
||||
@app.get("/items/{item_id}", tags=["items", "detail"])
|
||||
async def get_item(
|
||||
item_id: Annotated[str, Field(description="PATH_PARAM_DESCRIPTION")],
|
||||
fields: Annotated[
|
||||
str | None, Field(description="QUERY_PARAM_DESCRIPTION")
|
||||
] = None,
|
||||
) -> ItemResponse:
|
||||
"""FUNCTION_GET_DESCRIPTION
|
||||
|
||||
Gets a specific item by ID.
|
||||
|
||||
Args:
|
||||
item_id: The ID of the item to retrieve
|
||||
fields: Optional fields to include
|
||||
"""
|
||||
return ItemResponse(
|
||||
id=item_id, name=f"Item {item_id}", price=float(item_id) * 10.0
|
||||
)
|
||||
|
||||
@app.post("/items", tags=["items", "create"])
|
||||
async def create_item(item: Item) -> ItemResponse:
|
||||
"""FUNCTION_CREATE_DESCRIPTION
|
||||
|
||||
Creates a new item.
|
||||
|
||||
Body:
|
||||
Item object with name and price
|
||||
"""
|
||||
return ItemResponse(id="new", name=item.name, price=item.price)
|
||||
|
||||
return app
|
||||
|
||||
@pytest.fixture
|
||||
async def fastapi_server(self, fastapi_app_with_descriptions):
|
||||
"""Create a FastMCP server from the FastAPI app with custom route mappings."""
|
||||
# First create from FastAPI app to get the OpenAPI spec
|
||||
openapi_spec = fastapi_app_with_descriptions.openapi()
|
||||
|
||||
# Debug: check the operationIds in the OpenAPI spec
|
||||
print("\nDEBUG - OpenAPI Paths:")
|
||||
for path, methods in openapi_spec["paths"].items():
|
||||
for method, details in methods.items():
|
||||
if method != "parameters": # Skip non-HTTP method keys
|
||||
operation_id = details.get("operationId", "no_operation_id")
|
||||
print(
|
||||
f" Path: {path}, Method: {method}, OperationId: {operation_id}"
|
||||
)
|
||||
|
||||
# Create custom route mappings
|
||||
route_maps = [
|
||||
# Map GET /items to Resource
|
||||
RouteMap(
|
||||
methods=["GET"], pattern=r"^/items$", route_type=RouteType.RESOURCE
|
||||
),
|
||||
# Map GET /items/{item_id} to ResourceTemplate
|
||||
RouteMap(
|
||||
methods=["GET"],
|
||||
pattern=r"^/items/\{.*\}$",
|
||||
route_type=RouteType.RESOURCE_TEMPLATE,
|
||||
),
|
||||
# Map POST /items to Tool
|
||||
RouteMap(methods=["POST"], pattern=r"^/items$", route_type=RouteType.TOOL),
|
||||
]
|
||||
|
||||
# Create FastMCP server with the OpenAPI spec and custom route mappings
|
||||
server = FastMCPOpenAPI(
|
||||
openapi_spec=openapi_spec,
|
||||
client=AsyncClient(
|
||||
transport=ASGITransport(app=fastapi_app_with_descriptions),
|
||||
base_url="http://test",
|
||||
),
|
||||
name="Test FastAPI App",
|
||||
route_maps=route_maps,
|
||||
)
|
||||
|
||||
# Debug: print all components created
|
||||
print("\nDEBUG - Resources created:")
|
||||
for name, resource in server._resource_manager.get_resources().items():
|
||||
print(f" Resource: {name}, Name attribute: {resource.name}")
|
||||
|
||||
print("\nDEBUG - Templates created:")
|
||||
for name, template in server._resource_manager.get_templates().items():
|
||||
print(f" Template: {name}, Name attribute: {template.name}")
|
||||
|
||||
print("\nDEBUG - Tools created:")
|
||||
for tool in server._tool_manager.list_tools():
|
||||
print(f" Tool: {tool.name}")
|
||||
|
||||
return server
|
||||
|
||||
async def test_resource_includes_function_docstring(self, fastapi_server):
|
||||
"""Test that a Resource includes the function docstring."""
|
||||
resources = list(fastapi_server._resource_manager.get_resources().values())
|
||||
|
||||
# Now checking for the get_items operation ID rather than list_items
|
||||
list_resource = next((r for r in resources if "items_get" in r.name), None)
|
||||
|
||||
assert list_resource is not None, "GET /items resource wasn't created"
|
||||
description = list_resource.description or ""
|
||||
assert "FUNCTION_LIST_DESCRIPTION" in description, (
|
||||
"Function docstring missing from Resource"
|
||||
)
|
||||
|
||||
async def test_resource_includes_response_model_fields(self, fastapi_server):
|
||||
"""Test that a Resource description includes basic response information.
|
||||
|
||||
Note: FastAPI doesn't reliably include Pydantic field descriptions in the OpenAPI schema,
|
||||
so we can only check for basic response information being present.
|
||||
"""
|
||||
resources = list(fastapi_server._resource_manager.get_resources().values())
|
||||
list_resource = next((r for r in resources if "items_get" in r.name), None)
|
||||
|
||||
assert list_resource is not None, "GET /items resource wasn't created"
|
||||
description = list_resource.description or ""
|
||||
|
||||
# Check that at least the response information is included
|
||||
assert "Successful Response" in description, (
|
||||
"Response information missing from Resource description"
|
||||
)
|
||||
|
||||
# We've already verified in TestDescriptionPropagation that when descriptions
|
||||
# are present in the OpenAPI schema, they are properly included in the component description
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
assert get_template is not None, "GET /items/{item_id} template wasn't created"
|
||||
description = get_template.description or ""
|
||||
assert "FUNCTION_GET_DESCRIPTION" in description, (
|
||||
"Function docstring missing from ResourceTemplate"
|
||||
)
|
||||
|
||||
async def test_template_includes_path_parameter_description(self, fastapi_server):
|
||||
"""Test that a ResourceTemplate includes path parameter descriptions.
|
||||
|
||||
Note: Currently, FastAPI parameter descriptions using Annotated[type, Field(description=...)]
|
||||
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
|
||||
)
|
||||
|
||||
assert get_template is not None, "GET /items/{item_id} template wasn't created"
|
||||
description = get_template.description or ""
|
||||
|
||||
# Just test that parameters are included at all
|
||||
assert "Path Parameters" in description, (
|
||||
"Path parameters section missing from ResourceTemplate description"
|
||||
)
|
||||
assert "item_id" in description, (
|
||||
"item_id parameter missing from ResourceTemplate description"
|
||||
)
|
||||
|
||||
async def test_template_includes_query_parameter_description(self, fastapi_server):
|
||||
"""Test that a ResourceTemplate includes query parameter descriptions.
|
||||
|
||||
Note: Currently, FastAPI parameter descriptions using Annotated[type, Field(description=...)]
|
||||
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
|
||||
)
|
||||
|
||||
assert get_template is not None, "GET /items/{item_id} template wasn't created"
|
||||
description = get_template.description or ""
|
||||
|
||||
# Just test that parameters are included at all
|
||||
assert "Query Parameters" in description, (
|
||||
"Query parameters section missing from ResourceTemplate description"
|
||||
)
|
||||
assert "fields" in description, (
|
||||
"fields parameter missing from ResourceTemplate description"
|
||||
)
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
assert get_template is not None, "GET /items/{item_id} template wasn't created"
|
||||
assert "properties" in get_template.parameters, (
|
||||
"Schema properties missing from ResourceTemplate"
|
||||
)
|
||||
assert "item_id" in get_template.parameters["properties"], (
|
||||
"item_id missing from ResourceTemplate schema"
|
||||
)
|
||||
assert "description" in get_template.parameters["properties"]["item_id"], (
|
||||
"Description missing from item_id parameter schema"
|
||||
)
|
||||
assert (
|
||||
"PATH_PARAM_DESCRIPTION"
|
||||
in get_template.parameters["properties"]["item_id"]["description"]
|
||||
), "Path parameter description incorrect in schema"
|
||||
|
||||
async def test_tool_includes_function_docstring(self, fastapi_server):
|
||||
"""Test that a Tool includes the function docstring."""
|
||||
tools = fastapi_server._tool_manager.list_tools()
|
||||
create_tool = next(
|
||||
(t for t in tools if "create_item_items_post" == t.name), None
|
||||
)
|
||||
|
||||
assert create_tool is not None, "POST /items tool wasn't created"
|
||||
description = create_tool.description or ""
|
||||
assert "FUNCTION_CREATE_DESCRIPTION" in description, (
|
||||
"Function docstring missing from Tool"
|
||||
)
|
||||
|
||||
async def test_tool_parameter_schema_includes_property_description(
|
||||
self, fastapi_server
|
||||
):
|
||||
"""Test that a Tool's parameter schema includes property descriptions from request model.
|
||||
|
||||
Note: Currently, model field descriptions defined in Pydantic models using Field(description=...)
|
||||
may not be consistently propagated into the FastAPI OpenAPI schema and thus not into the tool's
|
||||
parameter schema.
|
||||
"""
|
||||
tools = fastapi_server._tool_manager.list_tools()
|
||||
create_tool = next(
|
||||
(t for t in tools if "create_item_items_post" == t.name), None
|
||||
)
|
||||
|
||||
assert create_tool is not None, "POST /items tool wasn't created"
|
||||
assert "properties" in create_tool.parameters, (
|
||||
"Schema properties missing from Tool"
|
||||
)
|
||||
assert "name" in create_tool.parameters["properties"], (
|
||||
"name parameter missing from Tool schema"
|
||||
)
|
||||
# We don't test for the description field content as it may not be consistently propagated
|
||||
|
||||
async def test_client_api_resource_description(self, fastapi_server):
|
||||
"""Test that Resource descriptions are accessible via the client API."""
|
||||
async with Client(fastapi_server) as client:
|
||||
resources = await client.list_resources()
|
||||
list_resource = next((r for r in resources if "items_get" in r.name), None)
|
||||
|
||||
assert list_resource is not None, (
|
||||
"GET /items resource not accessible via client API"
|
||||
)
|
||||
resource_description = list_resource.description or ""
|
||||
assert "FUNCTION_LIST_DESCRIPTION" in resource_description, (
|
||||
"Function docstring missing in Resource from client API"
|
||||
)
|
||||
|
||||
async def test_client_api_template_description(self, fastapi_server):
|
||||
"""Test that ResourceTemplate descriptions are accessible via the client API."""
|
||||
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
|
||||
)
|
||||
|
||||
assert get_template is not None, (
|
||||
"GET /items/{item_id} template not accessible via client API"
|
||||
)
|
||||
template_description = get_template.description or ""
|
||||
assert "FUNCTION_GET_DESCRIPTION" in template_description, (
|
||||
"Function docstring missing in ResourceTemplate from client API"
|
||||
)
|
||||
|
||||
async def test_client_api_tool_description(self, fastapi_server):
|
||||
"""Test that Tool descriptions are accessible via the client API."""
|
||||
async with Client(fastapi_server) as client:
|
||||
tools = await client.list_tools()
|
||||
create_tool = next(
|
||||
(t for t in tools if "create_item_items_post" == t.name), None
|
||||
)
|
||||
|
||||
assert create_tool is not None, (
|
||||
"POST /items tool not accessible via client API"
|
||||
)
|
||||
tool_description = create_tool.description or ""
|
||||
assert "FUNCTION_CREATE_DESCRIPTION" in tool_description, (
|
||||
"Function docstring missing in Tool from client API"
|
||||
)
|
||||
|
||||
async def test_client_api_tool_parameter_schema(self, fastapi_server):
|
||||
"""Test that Tool parameter schemas are accessible via the client API."""
|
||||
async with Client(fastapi_server) as client:
|
||||
tools = await client.list_tools()
|
||||
create_tool = next(
|
||||
(t for t in tools if "create_item_items_post" == t.name), None
|
||||
)
|
||||
|
||||
assert create_tool is not None, (
|
||||
"POST /items tool not accessible via client API"
|
||||
)
|
||||
assert "properties" in create_tool.inputSchema, (
|
||||
"Schema properties missing from Tool inputSchema in client API"
|
||||
)
|
||||
assert "name" in create_tool.inputSchema["properties"], (
|
||||
"name parameter missing from Tool schema in client API"
|
||||
)
|
||||
# We don't test for the description field content as it may not be consistently propagated
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue