mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-10 15:49:10 +02:00
b501f05794* Switch to new OpenAPI parser as default Remove the legacy OpenAPI parser and make the experimental parser the default. The experimental parser (introduced in 2.11) offers better performance, improved compatibility, and a more maintainable architecture. - Delete legacy parser (server/openapi.py, utilities/openapi.py) - Move experimental parser to main locations - Remove enable_new_openapi_parser feature flag - Update documentation to remove experimental references * Add deprecation stubs for experimental OpenAPI imports * Add deprecated enable_new_openapi_parser setting and deprecation tests * SDK docs * REview comments * Fix docstrings * Update docstring * Review comments * Fix broken links
337 lines
14 KiB
Python
337 lines
14 KiB
Python
"""Unit tests for FastMCPOpenAPI server."""
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from fastmcp.client import Client
|
|
from fastmcp.server.openapi import FastMCPOpenAPI
|
|
|
|
|
|
class TestFastMCPOpenAPIBasicFunctionality:
|
|
"""Test basic FastMCPOpenAPI server functionality."""
|
|
|
|
@pytest.fixture
|
|
def simple_openapi_spec(self):
|
|
"""Simple OpenAPI spec for testing."""
|
|
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 user by ID",
|
|
"parameters": [
|
|
{
|
|
"name": "id",
|
|
"in": "path",
|
|
"required": True,
|
|
"schema": {"type": "integer"},
|
|
}
|
|
],
|
|
"responses": {
|
|
"200": {
|
|
"description": "User retrieved successfully",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "integer"},
|
|
"name": {"type": "string"},
|
|
"email": {"type": "string"},
|
|
},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
}
|
|
},
|
|
"/users": {
|
|
"post": {
|
|
"operationId": "create_user",
|
|
"summary": "Create a new user",
|
|
"requestBody": {
|
|
"required": True,
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string"},
|
|
"email": {"type": "string"},
|
|
},
|
|
"required": ["name", "email"],
|
|
}
|
|
}
|
|
},
|
|
},
|
|
"responses": {
|
|
"201": {
|
|
"description": "User created successfully",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "integer"},
|
|
"name": {"type": "string"},
|
|
"email": {"type": "string"},
|
|
},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
}
|
|
},
|
|
},
|
|
}
|
|
|
|
def test_server_initialization(self, simple_openapi_spec):
|
|
"""Test server initialization with OpenAPI spec."""
|
|
client = httpx.AsyncClient(base_url="https://api.example.com")
|
|
|
|
server = FastMCPOpenAPI(
|
|
openapi_spec=simple_openapi_spec, client=client, name="Test Server"
|
|
)
|
|
|
|
assert server.name == "Test Server"
|
|
# Should have initialized RequestDirector successfully
|
|
assert hasattr(server, "_director")
|
|
assert hasattr(server, "_spec")
|
|
|
|
def test_server_initialization_with_custom_name(self, simple_openapi_spec):
|
|
"""Test server initialization with custom name."""
|
|
client = httpx.AsyncClient(base_url="https://api.example.com")
|
|
|
|
server = FastMCPOpenAPI(openapi_spec=simple_openapi_spec, client=client)
|
|
|
|
# Should use default name
|
|
assert server.name == "OpenAPI FastMCP"
|
|
|
|
async def test_server_creates_tools_from_spec(self, simple_openapi_spec):
|
|
"""Test that server creates tools from OpenAPI spec."""
|
|
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
|
|
server = FastMCPOpenAPI(
|
|
openapi_spec=simple_openapi_spec, client=client, name="Test Server"
|
|
)
|
|
|
|
# Test with in-memory client
|
|
async with Client(server) as mcp_client:
|
|
tools = await mcp_client.list_tools()
|
|
|
|
# Should have created tools for both operations
|
|
assert len(tools) == 2
|
|
|
|
tool_names = {tool.name for tool in tools}
|
|
assert "get_user" in tool_names
|
|
assert "create_user" in tool_names
|
|
|
|
async def test_server_tool_execution_fallback_to_http(self, simple_openapi_spec):
|
|
"""Test tool execution falls back to HTTP when callables aren't available."""
|
|
# Use a mock client that will be used for HTTP fallback
|
|
mock_client = httpx.AsyncClient()
|
|
|
|
server = FastMCPOpenAPI(
|
|
openapi_spec=simple_openapi_spec, client=mock_client, name="Test Server"
|
|
)
|
|
|
|
# With new architecture, tools are always created using RequestDirector
|
|
|
|
async with Client(server) as mcp_client:
|
|
tools = await mcp_client.list_tools()
|
|
|
|
# Should still have tools even without callables
|
|
assert len(tools) == 2
|
|
|
|
# Tools should be OpenAPITool instances using RequestDirector
|
|
# We'll just verify they exist and are callable
|
|
get_user_tool = next(tool for tool in tools if tool.name == "get_user")
|
|
assert get_user_tool is not None
|
|
assert get_user_tool.description is not None
|
|
|
|
def test_server_request_director_initialization(self, simple_openapi_spec):
|
|
"""Test that server initializes RequestDirector successfully."""
|
|
client = httpx.AsyncClient(base_url="https://api.example.com")
|
|
|
|
# This should not raise an exception
|
|
server = FastMCPOpenAPI(
|
|
openapi_spec=simple_openapi_spec, client=client, name="Test Server"
|
|
)
|
|
|
|
# Server should be created successfully
|
|
assert server is not None
|
|
assert server.name == "Test Server"
|
|
# RequestDirector and Spec should be initialized
|
|
assert hasattr(server, "_director")
|
|
assert hasattr(server, "_spec")
|
|
|
|
def test_server_with_timeout(self, simple_openapi_spec):
|
|
"""Test server initialization with timeout setting."""
|
|
client = httpx.AsyncClient(base_url="https://api.example.com")
|
|
|
|
server = FastMCPOpenAPI(
|
|
openapi_spec=simple_openapi_spec,
|
|
client=client,
|
|
name="Test Server",
|
|
timeout=30.0,
|
|
)
|
|
|
|
assert server._timeout == 30.0
|
|
|
|
def test_server_with_empty_spec(self):
|
|
"""Test server with minimal OpenAPI spec."""
|
|
minimal_spec = {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Empty API", "version": "1.0.0"},
|
|
"paths": {},
|
|
}
|
|
|
|
client = httpx.AsyncClient(base_url="https://api.example.com")
|
|
|
|
server = FastMCPOpenAPI(
|
|
openapi_spec=minimal_spec, client=client, name="Empty Server"
|
|
)
|
|
|
|
assert server.name == "Empty Server"
|
|
# Should handle empty paths gracefully
|
|
assert hasattr(server, "_director")
|
|
assert hasattr(server, "_spec")
|
|
|
|
async def test_clean_schema_output_no_unused_defs(self):
|
|
"""Test that unused schema definitions are removed from tool schemas."""
|
|
# Create a spec with unused HTTPValidationError-like definitions
|
|
spec_with_unused_defs = {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Test API", "version": "1.0.0"},
|
|
"servers": [{"url": "https://api.example.com"}],
|
|
"paths": {
|
|
"/users": {
|
|
"post": {
|
|
"operationId": "create_user",
|
|
"summary": "Create a new user",
|
|
"requestBody": {
|
|
"required": True,
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string", "title": "Name"},
|
|
"active": {
|
|
"type": "boolean",
|
|
"title": "Active",
|
|
},
|
|
},
|
|
"required": ["name", "active"],
|
|
}
|
|
}
|
|
},
|
|
},
|
|
"responses": {
|
|
"200": {
|
|
"description": "User created successfully",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {
|
|
"type": "integer",
|
|
"title": "Id",
|
|
},
|
|
"name": {
|
|
"type": "string",
|
|
"title": "Name",
|
|
},
|
|
"active": {
|
|
"type": "boolean",
|
|
"title": "Active",
|
|
},
|
|
},
|
|
"required": ["id", "name", "active"],
|
|
"title": "User",
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
}
|
|
}
|
|
},
|
|
"components": {
|
|
"schemas": {
|
|
# This should be removed since it's not referenced
|
|
"HTTPValidationError": {
|
|
"properties": {
|
|
"detail": {
|
|
"items": {
|
|
"$ref": "#/components/schemas/ValidationError"
|
|
},
|
|
"title": "Detail",
|
|
"type": "array",
|
|
}
|
|
},
|
|
"title": "HTTPValidationError",
|
|
"type": "object",
|
|
},
|
|
"ValidationError": {
|
|
"properties": {
|
|
"loc": {
|
|
"items": {
|
|
"anyOf": [{"type": "string"}, {"type": "integer"}]
|
|
},
|
|
"title": "Location",
|
|
"type": "array",
|
|
},
|
|
"msg": {"title": "Message", "type": "string"},
|
|
"type": {"title": "Error Type", "type": "string"},
|
|
},
|
|
"required": ["loc", "msg", "type"],
|
|
"title": "ValidationError",
|
|
"type": "object",
|
|
},
|
|
}
|
|
},
|
|
}
|
|
|
|
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
|
|
server = FastMCPOpenAPI(
|
|
openapi_spec=spec_with_unused_defs, client=client, name="Test Server"
|
|
)
|
|
|
|
async with Client(server) as mcp_client:
|
|
tools = await mcp_client.list_tools()
|
|
|
|
assert len(tools) == 1 # Only the POST operation
|
|
tool = tools[0]
|
|
|
|
# Verify tool has clean schemas without unused $defs
|
|
assert tool.name == "create_user"
|
|
|
|
# Input schema should not have $defs since no references are used
|
|
expected_input_schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string", "title": "Name"},
|
|
"active": {"type": "boolean", "title": "Active"},
|
|
},
|
|
"required": ["name", "active"],
|
|
}
|
|
assert tool.inputSchema == expected_input_schema
|
|
|
|
# Output schema should not have $defs since no references are used
|
|
expected_output_schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "integer", "title": "Id"},
|
|
"name": {"type": "string", "title": "Name"},
|
|
"active": {"type": "boolean", "title": "Active"},
|
|
},
|
|
"required": ["id", "name", "active"],
|
|
"title": "User",
|
|
}
|
|
assert tool.outputSchema == expected_output_schema
|