"""Tests for OpenAPI feature support in OpenAPIProvider.""" import json from typing import Any from unittest.mock import AsyncMock, Mock import httpx2 import pytest from httpx2 import Response from mcp_types import TextResourceContents 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, _redact_headers, ) from fastmcp.server.providers.openapi.routing import MCPType, RouteMap from fastmcp.utilities.openapi.models import HTTPRoute, ResponseInfo # Real client used only to delegate build_request on mocked clients - never sends. _request_builder = httpx2.AsyncClient() def create_openapi_server( openapi_spec: dict, client, name: str = "OpenAPI Server", ) -> FastMCP: """Helper to create a FastMCP server with OpenAPIProvider.""" provider = OpenAPIProvider(openapi_spec=openapi_spec, client=client) mcp = FastMCP(name) mcp.add_provider(provider) return mcp class TestParameterHandling: """Test OpenAPI parameter handling features.""" @pytest.fixture def parameter_spec(self): """OpenAPI spec with various parameter types.""" return { "openapi": "3.0.0", "info": {"title": "Parameter Test API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com"}], "paths": { "/search": { "get": { "operationId": "search_items", "summary": "Search items", "parameters": [ { "name": "query", "in": "query", "required": True, "schema": {"type": "string"}, "description": "Search query", }, { "name": "limit", "in": "query", "required": False, "schema": { "type": "integer", "minimum": 1, "maximum": 100, }, "description": "Maximum number of results", }, { "name": "tags", "in": "query", "required": False, "schema": { "type": "array", "items": {"type": "string"}, }, "style": "form", "explode": True, "description": "Filter by tags", }, { "name": "X-API-Key", "in": "header", "required": True, "schema": {"type": "string"}, "description": "API key for authentication", }, ], "responses": { "200": { "description": "Search results", "content": { "application/json": { "schema": { "type": "object", "properties": { "items": { "type": "array", "items": {"type": "object"}, }, "total": {"type": "integer"}, }, } } }, } }, } }, "/users/{id}/posts/{post_id}": { "get": { "operationId": "get_user_post", "summary": "Get specific user post", "parameters": [ { "name": "id", "in": "path", "required": True, "schema": {"type": "integer"}, "description": "User ID", }, { "name": "post_id", "in": "path", "required": True, "schema": {"type": "integer"}, "description": "Post ID", }, ], "responses": { "200": { "description": "User post", "content": { "application/json": { "schema": { "type": "object", "properties": { "id": {"type": "integer"}, "title": {"type": "string"}, "content": {"type": "string"}, }, } } }, } }, } }, }, } async def test_query_parameters_in_tools(self, parameter_spec): """Test that query parameters are properly included in tool parameters.""" async with httpx2.AsyncClient(base_url="https://api.example.com") as client: server = create_openapi_server( openapi_spec=parameter_spec, client=client, name="Parameter Test Server" ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() # Find the search tool search_tool = next( tool for tool in tools if tool.name == "search_items" ) assert search_tool is not None # Check that parameters are included in the tool's input schema params = search_tool.input_schema assert params["type"] == "object" properties = params["properties"] # Check that key parameters are present # (Schema details may vary based on implementation) assert "query" in properties assert "limit" in properties assert "tags" in properties assert "X-API-Key" in properties # Check that parameter descriptions are included assert "description" in properties["query"], ( "Query parameter should have description" ) assert properties["query"]["description"] == "Search query" assert "description" in properties["limit"], ( "Limit parameter should have description" ) assert properties["limit"]["description"] == "Maximum number of results" assert "description" in properties["tags"], ( "Tags parameter should have description" ) assert properties["tags"]["description"] == "Filter by tags" # Check that required parameters are marked as required required = params.get("required", []) assert "query" in required assert "X-API-Key" in required async def test_path_parameters_in_tools(self, parameter_spec): """Test that path parameters are properly included in tool parameters.""" async with httpx2.AsyncClient(base_url="https://api.example.com") as client: server = create_openapi_server( openapi_spec=parameter_spec, client=client, name="Parameter Test Server" ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() # Find the user post tool user_post_tool = next( tool for tool in tools if tool.name == "get_user_post" ) assert user_post_tool is not None # Check that path parameters are included params = user_post_tool.input_schema properties = params["properties"] # Check that path parameters are present assert "id" in properties assert "post_id" in properties # Path parameters should be required required = params.get("required", []) assert "id" in required assert "post_id" in required class TestRequestBodyHandling: """Test OpenAPI request body handling.""" @pytest.fixture def request_body_spec(self): """OpenAPI spec with request body.""" return { "openapi": "3.0.0", "info": {"title": "Request Body Test API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com"}], "paths": { "/users": { "post": { "operationId": "create_user", "summary": "Create a user", "requestBody": { "required": True, "content": { "application/json": { "schema": { "type": "object", "properties": { "name": { "type": "string", "description": "User's full name", }, "email": { "type": "string", "format": "email", "description": "User's email address", }, "age": { "type": "integer", "minimum": 0, "maximum": 150, "description": "User's age", }, "preferences": { "type": "object", "properties": { "theme": {"type": "string"}, "notifications": { "type": "boolean" }, }, "description": "User preferences", }, }, "required": ["name", "email"], } } }, }, "responses": { "201": { "description": "User created", "content": { "application/json": { "schema": { "type": "object", "properties": { "id": {"type": "integer"}, "name": {"type": "string"}, "email": {"type": "string"}, }, } } }, } }, } } }, } async def test_request_body_properties_in_tool(self, request_body_spec): """Test that request body properties are included in tool parameters.""" async with httpx2.AsyncClient(base_url="https://api.example.com") as client: server = create_openapi_server( openapi_spec=request_body_spec, client=client, name="Request Body Test Server", ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() # Find the create user tool create_tool = next(tool for tool in tools if tool.name == "create_user") assert create_tool is not None # Check that request body properties are included params = create_tool.input_schema properties = params["properties"] # Check that request body properties are present assert "name" in properties assert "email" in properties assert "age" in properties assert "preferences" in properties # Check required fields from request body required = params.get("required", []) assert "name" in required assert "email" in required class TestResponseSchemas: """Test OpenAPI response schema handling.""" @pytest.fixture def response_schema_spec(self): """OpenAPI spec with detailed response schemas.""" return { "openapi": "3.0.0", "info": {"title": "Response Schema Test API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com"}], "paths": { "/users/{id}": { "get": { "operationId": "get_user", "summary": "Get user details", "parameters": [ { "name": "id", "in": "path", "required": True, "schema": {"type": "integer"}, } ], "responses": { "200": { "description": "User details retrieved successfully", "content": { "application/json": { "schema": { "type": "object", "properties": { "id": {"type": "integer"}, "name": {"type": "string"}, "email": {"type": "string"}, "profile": { "type": "object", "properties": { "bio": {"type": "string"}, "avatar_url": { "type": "string" }, }, }, }, "required": ["id", "name", "email"], } } }, }, "404": { "description": "User not found", "content": { "application/json": { "schema": { "type": "object", "properties": { "error": {"type": "string"}, "code": {"type": "integer"}, }, } } }, }, }, } } }, } async def test_tool_has_output_schema(self, response_schema_spec): """Test that tools have output schemas from response definitions.""" async with httpx2.AsyncClient(base_url="https://api.example.com") as client: server = create_openapi_server( openapi_spec=response_schema_spec, client=client, name="Response Schema Test Server", ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() # Find the get user tool get_user_tool = next(tool for tool in tools if tool.name == "get_user") assert get_user_tool is not None # Check that the tool has an output schema # Note: output schema might be None if not extracted properly # 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 httpx2.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].mime_type == "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 httpx2.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].mime_type == "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 httpx2.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].mime_type == "application/json" class TestResourceTemplateRequestBuilding: @pytest.fixture def path_param_spec(self) -> dict[str, Any]: return { "openapi": "3.0.0", "info": {"title": "User API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com/api/v1"}], "paths": { "/users/{id}": { "get": { "operationId": "get_user", "summary": "Get user", "parameters": [ { "name": "id", "in": "path", "required": True, "schema": {"type": "string"}, } ], "responses": { "200": { "description": "User data", "content": { "application/json": {"schema": {"type": "object"}} }, } }, } } }, } async def test_resource_template_encodes_matched_path_params( self, path_param_spec: dict[str, Any] ): seen_urls: list[httpx2.URL] = [] async def handler(request: httpx2.Request) -> httpx2.Response: seen_urls.append(request.url) return httpx2.Response(200, json={"ok": True}) route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] async with httpx2.AsyncClient( base_url="https://api.example.com/api/v1", transport=httpx2.MockTransport(handler), ) as client: provider = OpenAPIProvider( openapi_spec=path_param_spec, client=client, route_maps=route_maps, ) mcp = FastMCP("Test") mcp.add_provider(provider) async with Client(mcp) as mcp_client: # Reserved characters (encoded slash + space) must be # re-encoded when building the outbound URL. A traversal # payload (`..%2F...`) would be rejected by the default # resource-security screening, so use a benign value that # still exercises reserved-character encoding. await mcp_client.read_resource("resource://get_user/a%2Fb%20c") assert seen_urls == [ httpx2.URL("https://api.example.com/api/v1/users/a%2Fb%20c") ] async def test_resource_template_ignores_unmatched_query_string( self, path_param_spec: dict[str, Any] ): seen_urls: list[httpx2.URL] = [] async def handler(request: httpx2.Request) -> httpx2.Response: seen_urls.append(request.url) return httpx2.Response(200, json={"ok": True}) route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] async with httpx2.AsyncClient( base_url="https://api.example.com/api/v1", transport=httpx2.MockTransport(handler), ) as client: provider = OpenAPIProvider( openapi_spec=path_param_spec, client=client, route_maps=route_maps, ) mcp = FastMCP("Test") mcp.add_provider(provider) async with Client(mcp) as mcp_client: await mcp_client.read_resource("resource://get_user/alice?admin=true") assert seen_urls == [httpx2.URL("https://api.example.com/api/v1/users/alice")] async def test_resource_template_preserves_hyphenated_path_params(self): seen_urls: list[httpx2.URL] = [] async def handler(request: httpx2.Request) -> httpx2.Response: seen_urls.append(request.url) return httpx2.Response(200, json={"ok": True}) spec = { "openapi": "3.0.0", "info": {"title": "User API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com/api/v1"}], "paths": { "/users/{user-id}": { "get": { "operationId": "get_user", "parameters": [ { "name": "user-id", "in": "path", "required": True, "schema": {"type": "string"}, } ], "responses": { "200": { "description": "User data", "content": { "application/json": {"schema": {"type": "object"}} }, } }, } } }, } route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] async with httpx2.AsyncClient( base_url="https://api.example.com/api/v1", transport=httpx2.MockTransport(handler), ) 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: await mcp_client.read_resource("resource://get_user/abc") assert seen_urls == [httpx2.URL("https://api.example.com/api/v1/users/abc")] async def test_resource_template_preserves_client_defaults( self, path_param_spec: dict[str, Any] ): seen_requests: list[httpx2.Request] = [] async def handler(request: httpx2.Request) -> httpx2.Response: seen_requests.append(request) return httpx2.Response(200, json={"ok": True}) route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] async with httpx2.AsyncClient( base_url="https://api.example.com/api/v1", params={"api-version": "2026-06-29"}, cookies={"session": "abc123"}, transport=httpx2.MockTransport(handler), ) as client: provider = OpenAPIProvider( openapi_spec=path_param_spec, client=client, route_maps=route_maps, ) mcp = FastMCP("Test") mcp.add_provider(provider) async with Client(mcp) as mcp_client: await mcp_client.read_resource("resource://get_user/alice") assert seen_requests[0].url == httpx2.URL( "https://api.example.com/api/v1/users/alice?api-version=2026-06-29" ) assert seen_requests[0].headers["cookie"] == "session=abc123" async def test_resource_template_uses_mapped_path_argument_names(self): seen_urls: list[httpx2.URL] = [] async def handler(request: httpx2.Request) -> httpx2.Response: seen_urls.append(request.url) return httpx2.Response(200, json={"ok": True}) spec = { "openapi": "3.0.0", "info": {"title": "User API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com/api/v1"}], "paths": { "/users/{id}": { "get": { "operationId": "get_user", "parameters": [ { "name": "id", "in": "path", "required": True, "schema": {"type": "string"}, } ], "requestBody": { "content": { "application/json": { "schema": { "type": "object", "properties": { "id": {"type": "string"}, "name": {"type": "string"}, }, } } } }, "responses": { "200": { "description": "User data", "content": { "application/json": {"schema": {"type": "object"}} }, } }, } } }, } route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] async with httpx2.AsyncClient( base_url="https://api.example.com/api/v1", transport=httpx2.MockTransport(handler), ) 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: await mcp_client.read_resource("resource://get_user/abc") assert seen_urls == [httpx2.URL("https://api.example.com/api/v1/users/abc")] async def test_resource_template_uses_path_arg_when_query_param_has_same_name( self, ): seen_urls: list[httpx2.URL] = [] async def handler(request: httpx2.Request) -> httpx2.Response: seen_urls.append(request.url) return httpx2.Response(200, json={"ok": True}) spec = { "openapi": "3.0.0", "info": {"title": "User API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com/api/v1"}], "paths": { "/users/{id}": { "get": { "operationId": "get_user", "parameters": [ { "name": "id", "in": "path", "required": True, "schema": {"type": "string"}, }, { "name": "id", "in": "query", "required": False, "schema": {"type": "string"}, }, ], "responses": { "200": { "description": "User data", "content": { "application/json": {"schema": {"type": "object"}} }, } }, } } }, } route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE_TEMPLATE)] async with httpx2.AsyncClient( base_url="https://api.example.com/api/v1", transport=httpx2.MockTransport(handler), ) 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: await mcp_client.read_resource("resource://get_user/abc") assert seen_urls == [httpx2.URL("https://api.example.com/api/v1/users/abc")] 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 httpx2.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].mime_type == "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 httpx2.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].mime_type == "text/plain" async def test_resource_plus_json_response_is_text(self): """A +json response is read as JSON text, not a binary blob.""" spec = { "openapi": "3.0.0", "info": {"title": "Status API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com"}], "paths": { "/status": { "get": { "operationId": "get_status", "responses": { "200": { "description": "Status", "content": { "application/hal+json": { "schema": {"type": "object"} } }, } }, } } }, } def handler(request: httpx2.Request) -> httpx2.Response: return httpx2.Response( 200, content=b'{"ok": true}', headers={"content-type": "application/hal+json; charset=utf-8"}, ) route_maps = [RouteMap(methods=["GET"], mcp_type=MCPType.RESOURCE)] async with httpx2.AsyncClient( base_url="https://api.example.com", transport=httpx2.MockTransport(handler), ) 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: [content] = await mcp_client.read_resource("resource://get_status") assert isinstance(content, TextResourceContents) assert json.loads(content.text) == {"ok": True} assert content.mime_type == "application/hal+json" class TestValidateOutput: """Tests for the validate_output option on OpenAPIProvider.""" @pytest.fixture def spec_with_output_schema(self): return { "openapi": "3.0.0", "info": {"title": "Test API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com"}], "paths": { "/users/{id}": { "get": { "operationId": "get_user", "summary": "Get a user", "parameters": [ { "name": "id", "in": "path", "required": True, "schema": {"type": "integer"}, } ], "responses": { "200": { "description": "A user", "content": { "application/json": { "schema": { "type": "object", "properties": { "id": {"type": "integer"}, "name": {"type": "string"}, "email": {"type": "string"}, }, "required": ["id", "name"], } } }, } }, } }, "/items": { "get": { "operationId": "list_items", "summary": "List items", "responses": { "200": { "description": "An array of items", "content": { "application/json": { "schema": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"} }, }, } } }, } }, } }, }, } async def test_validate_output_true_preserves_extracted_schema( self, spec_with_output_schema ): """Default validate_output=True uses the real extracted schema.""" async with httpx2.AsyncClient(base_url="https://api.example.com") as client: provider = OpenAPIProvider( openapi_spec=spec_with_output_schema, client=client, ) tool = provider._tools["get_user"] assert tool.output_schema is not None assert tool.output_schema.get("type") == "object" assert "properties" in tool.output_schema assert "id" in tool.output_schema["properties"] async def test_validate_output_false_uses_permissive_schema( self, spec_with_output_schema ): """validate_output=False replaces the schema with a permissive one.""" async with httpx2.AsyncClient(base_url="https://api.example.com") as client: provider = OpenAPIProvider( openapi_spec=spec_with_output_schema, client=client, validate_output=False, ) tool = provider._tools["get_user"] assert tool.output_schema is not None assert tool.output_schema == { "type": "object", "additionalProperties": True, } async def test_validate_output_false_preserves_wrap_result_flag( self, spec_with_output_schema ): """validate_output=False preserves x-fastmcp-wrap-result for array responses.""" async with httpx2.AsyncClient(base_url="https://api.example.com") as client: provider = OpenAPIProvider( openapi_spec=spec_with_output_schema, client=client, validate_output=False, ) # The list_items endpoint returns an array, so the extracted schema # would have had x-fastmcp-wrap-result=True tool = provider._tools["list_items"] assert tool.output_schema is not None assert tool.output_schema.get("x-fastmcp-wrap-result") is True assert tool.output_schema.get("additionalProperties") is True async def test_validate_output_false_allows_nonconforming_response( self, spec_with_output_schema ): """With validate_output=False, responses that don't match the spec succeed.""" mock_client = Mock(spec=httpx2.AsyncClient) mock_client.base_url = "https://api.example.com" mock_client.headers = None mock_client.build_request = _request_builder.build_request # Return extra fields not in the schema mock_response = Mock(spec=Response) mock_response.status_code = 200 mock_response.json.return_value = { "id": 1, "name": "Alice", "email": "alice@example.com", "unexpected_field": "surprise", "nested": {"deep": True}, } mock_response.raise_for_status = Mock() mock_client.send = AsyncMock(return_value=mock_response) provider = OpenAPIProvider( openapi_spec=spec_with_output_schema, client=mock_client, validate_output=False, ) mcp = FastMCP("Test") mcp.add_provider(provider) async with Client(mcp) as mcp_client: result = await mcp_client.call_tool("get_user", {"id": 1}) assert result is not None # Structured content should have the full response including extra fields assert result.structured_content is not None assert result.structured_content["unexpected_field"] == "surprise" async def test_validate_output_false_wraps_non_dict_response( self, spec_with_output_schema ): """Non-dict responses are wrapped even when schema says object and validate_output=False.""" mock_client = Mock(spec=httpx2.AsyncClient) mock_client.base_url = "https://api.example.com" mock_client.headers = None mock_client.build_request = _request_builder.build_request # Backend returns an array even though schema says object mock_response = Mock(spec=Response) mock_response.status_code = 200 mock_response.json.return_value = [{"id": 1}, {"id": 2}] mock_response.raise_for_status = Mock() mock_client.send = AsyncMock(return_value=mock_response) provider = OpenAPIProvider( openapi_spec=spec_with_output_schema, client=mock_client, validate_output=False, ) mcp = FastMCP("Test") mcp.add_provider(provider) async with Client(mcp) as mcp_client: result = await mcp_client.call_tool("get_user", {"id": 1}) assert result is not None # Non-dict should be wrapped so structured_content is always a dict assert result.structured_content is not None assert isinstance(result.structured_content, dict) assert result.structured_content["result"] == [{"id": 1}, {"id": 2}] async def test_from_openapi_threads_validate_output(self, spec_with_output_schema): """FastMCP.from_openapi() correctly passes validate_output to the provider.""" mock_client = Mock(spec=httpx2.AsyncClient) mock_client.base_url = "https://api.example.com" mock_client.headers = None mock_client.build_request = _request_builder.build_request server = FastMCP.from_openapi( openapi_spec=spec_with_output_schema, client=mock_client, validate_output=False, ) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() get_user = next(t for t in tools if t.name == "get_user") # With validate_output=False, the outputSchema should be permissive assert get_user.output_schema is not None assert get_user.output_schema.get("additionalProperties") is True # Should NOT have specific properties from the original schema assert "properties" not in get_user.output_schema class TestRedactHeaders: """Test that non-safe headers are redacted in debug logging.""" def test_known_sensitive_headers_are_redacted(self): headers = httpx2.Headers( { "Authorization": "Bearer secret-token", "X-API-Key": "my-api-key", "Cookie": "session=abc123", "Proxy-Authorization": "Basic creds", "Content-Type": "application/json", "Accept": "text/html", } ) redacted = _redact_headers(headers) assert redacted["authorization"] == "***" assert redacted["x-api-key"] == "***" assert redacted["cookie"] == "***" assert redacted["proxy-authorization"] == "***" assert redacted["content-type"] == "application/json" assert redacted["accept"] == "text/html" def test_arbitrary_auth_headers_are_redacted(self): """Arbitrary header names (e.g. OpenAPI apiKey-in-header) are redacted.""" headers = httpx2.Headers( { "X-Custom-Token": "secret", "X-My-Service-Key": "also-secret", "Content-Type": "application/json", } ) redacted = _redact_headers(headers) assert redacted["x-custom-token"] == "***" assert redacted["x-my-service-key"] == "***" assert redacted["content-type"] == "application/json" def test_safe_only_headers(self): headers = httpx2.Headers({"Content-Type": "application/json"}) redacted = _redact_headers(headers) assert redacted == {"content-type": "application/json"} class TestMultipartUpload: """Multipart request bodies must survive the build_request rebuild. The director constructs multipart bodies with ``files=``, which yields a streaming request body; the rebuild through the user's client must materialize it (``read()``) rather than access ``.content``, which raises ``RequestNotRead`` on unread streams. """ MULTIPART_SPEC = { "openapi": "3.0.0", "info": {"title": "Upload API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com"}], "paths": { "/upload": { "post": { "operationId": "upload_file", "summary": "Upload a file", "requestBody": { "required": True, "content": { "multipart/form-data": { "schema": { "type": "object", "properties": {"file": {"type": "string"}}, } } }, }, "responses": { "200": { "description": "Uploaded", "content": { "application/json": { "schema": { "type": "object", "properties": {"ok": {"type": "boolean"}}, } } }, } }, } } }, } async def test_multipart_tool_call_sends_materialized_body(self): received: dict[str, Any] = {} def handler(request): received["content_type"] = request.headers.get("content-type", "") received["body"] = request.read() return httpx2.Response(200, json={"ok": True}) transport = httpx2.MockTransport(handler) async with httpx2.AsyncClient( transport=transport, base_url="https://api.example.com" ) as client: server = create_openapi_server(self.MULTIPART_SPEC, client) async with Client(server) as mcp_client: result = await mcp_client.call_tool("upload_file", {"file": "data"}) assert result.structured_content == {"ok": True} assert "multipart/form-data" in received["content_type"] assert b"data" in received["body"] class TestAllOfReferenceRequestBodies: """Request bodies keep fields inherited through an allOf reference.""" SPEC = { "openapi": "3.1.0", "info": {"title": "Pet API", "version": "1.0.0"}, "servers": [{"url": "https://api.example.com"}], "paths": { "/pets": { "post": { "operationId": "create_pet", "requestBody": { "required": True, "content": { "application/json": { "schema": {"$ref": "#/components/schemas/Cat"} } }, }, "responses": {"200": {"description": "Created"}}, } } }, "components": { "schemas": { "Animal": { "type": "object", "properties": {"animalId": {"type": "string"}}, "required": ["animalId"], }, "Pet": { "allOf": [ {"$ref": "#/components/schemas/Animal"}, { "type": "object", "properties": {"petType": {"type": "string"}}, "required": ["petType"], }, ] }, "Cat": { "allOf": [ {"$ref": "#/components/schemas/Pet"}, { "type": "object", "properties": {"meowVolume": {"type": "integer"}}, "required": ["meowVolume"], }, ] }, } }, } async def test_allof_reference_fields_reach_tool_schema_and_request_body(self): received: dict[str, object] = {} def handler(request): received["body"] = json.loads(request.content) return httpx2.Response(200, json={"ok": True}) async with httpx2.AsyncClient( transport=httpx2.MockTransport(handler), base_url="https://api.example.com", ) as client: server = create_openapi_server(self.SPEC, client) async with Client(server) as mcp_client: tools = await mcp_client.list_tools() tool = next(tool for tool in tools if tool.name == "create_pet") assert tool.input_schema["properties"].keys() >= { "animalId", "petType", "meowVolume", } result = await mcp_client.call_tool( "create_pet", {"animalId": "a-1", "petType": "cat", "meowVolume": 11}, ) assert result.structured_content == {"ok": True} assert received["body"] == { "animalId": "a-1", "petType": "cat", "meowVolume": 11, } async def test_allof_reference_request_body_does_not_crash(self): """Required fields inherited through a reference can be sent together.""" received: dict[str, object] = {} def handler(request): received["body"] = json.loads(request.content) return httpx2.Response(200, json={"ok": True}) async with httpx2.AsyncClient( transport=httpx2.MockTransport(handler), base_url="https://api.example.com", ) as client: server = create_openapi_server(self.SPEC, client) async with Client(server) as mcp_client: result = await mcp_client.call_tool( "create_pet", {"animalId": "a-1", "petType": "cat", "meowVolume": 11}, ) assert result.structured_content == {"ok": True} assert received["body"] == { "animalId": "a-1", "petType": "cat", "meowVolume": 11, }