mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-19 03:54:18 +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
325 lines
14 KiB
Python
325 lines
14 KiB
Python
"""End-to-end compatibility tests between legacy and new OpenAPI implementations."""
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from fastmcp.client import Client
|
|
from fastmcp.server.openapi import FastMCPOpenAPI
|
|
from fastmcp.server.openapi import FastMCPOpenAPI as LegacyFastMCPOpenAPI
|
|
|
|
|
|
class TestEndToEndCompatibility:
|
|
"""Test that legacy and new implementations create identical tools."""
|
|
|
|
@pytest.fixture
|
|
def simple_spec(self):
|
|
"""Simple OpenAPI spec for testing."""
|
|
return {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Test API", "version": "1.0.0"},
|
|
"paths": {
|
|
"/users/{id}": {
|
|
"get": {
|
|
"operationId": "get_user",
|
|
"summary": "Get user by ID",
|
|
"parameters": [
|
|
{
|
|
"name": "id",
|
|
"in": "path",
|
|
"required": True,
|
|
"schema": {"type": "integer"},
|
|
},
|
|
{
|
|
"name": "include_details",
|
|
"in": "query",
|
|
"required": False,
|
|
"schema": {"type": "boolean"},
|
|
},
|
|
],
|
|
"responses": {"200": {"description": "User found"}},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
@pytest.fixture
|
|
def collision_spec(self):
|
|
"""OpenAPI spec with parameter collisions."""
|
|
return {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Collision API", "version": "1.0.0"},
|
|
"paths": {
|
|
"/users/{id}": {
|
|
"put": {
|
|
"operationId": "update_user",
|
|
"summary": "Update user",
|
|
"parameters": [
|
|
{
|
|
"name": "id",
|
|
"in": "path",
|
|
"required": True,
|
|
"schema": {"type": "integer"},
|
|
}
|
|
],
|
|
"requestBody": {
|
|
"required": True,
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "integer"},
|
|
"name": {"type": "string"},
|
|
},
|
|
"required": ["name"],
|
|
}
|
|
}
|
|
},
|
|
},
|
|
"responses": {"200": {"description": "User updated"}},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
async def test_tool_schema_compatibility(self, simple_spec):
|
|
"""Test that tools have identical input schemas."""
|
|
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
|
|
# Create both servers
|
|
legacy_server = LegacyFastMCPOpenAPI(
|
|
openapi_spec=simple_spec,
|
|
client=client,
|
|
name="Legacy Server",
|
|
)
|
|
new_server = FastMCPOpenAPI(
|
|
openapi_spec=simple_spec,
|
|
client=client,
|
|
name="New Server",
|
|
)
|
|
|
|
# Get tools from both servers
|
|
async with Client(legacy_server) as legacy_client:
|
|
legacy_tools = await legacy_client.list_tools()
|
|
|
|
async with Client(new_server) as new_client:
|
|
new_tools = await new_client.list_tools()
|
|
|
|
# Should have same number of tools
|
|
assert len(legacy_tools) == len(new_tools)
|
|
assert len(legacy_tools) == 1
|
|
|
|
# Get the single tool from each
|
|
legacy_tool = legacy_tools[0]
|
|
new_tool = new_tools[0]
|
|
|
|
# Names should be identical
|
|
assert legacy_tool.name == new_tool.name
|
|
assert legacy_tool.name == "get_user"
|
|
|
|
# Descriptions may differ (new server has simplified descriptions)
|
|
# Just check that both have descriptions
|
|
assert legacy_tool.description
|
|
assert new_tool.description
|
|
|
|
# Input schemas should be identical
|
|
legacy_schema = legacy_tool.inputSchema
|
|
new_schema = new_tool.inputSchema
|
|
|
|
# Required fields should match
|
|
assert set(legacy_schema.get("required", [])) == set(
|
|
new_schema.get("required", [])
|
|
)
|
|
|
|
# Properties should match
|
|
legacy_props = legacy_schema.get("properties", {})
|
|
new_props = new_schema.get("properties", {})
|
|
assert set(legacy_props.keys()) == set(new_props.keys())
|
|
|
|
# Check each property
|
|
for prop_name in legacy_props:
|
|
legacy_prop = legacy_props[prop_name]
|
|
new_prop = new_props[prop_name]
|
|
|
|
# For required parameters, should have simple type
|
|
if prop_name in legacy_schema.get("required", []):
|
|
assert legacy_prop.get("type") == new_prop.get("type")
|
|
assert "anyOf" not in legacy_prop
|
|
assert "anyOf" not in new_prop
|
|
else:
|
|
# Both implementations now correctly preserve original schema without nullable behavior
|
|
assert "anyOf" not in legacy_prop
|
|
assert "anyOf" not in new_prop
|
|
# Both should have the same type
|
|
assert legacy_prop.get("type") == new_prop.get("type")
|
|
|
|
async def test_collision_handling_compatibility(self, collision_spec):
|
|
"""Test that parameter collision handling is identical."""
|
|
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
|
|
# Create both servers
|
|
legacy_server = LegacyFastMCPOpenAPI(
|
|
openapi_spec=collision_spec,
|
|
client=client,
|
|
name="Legacy Server",
|
|
)
|
|
new_server = FastMCPOpenAPI(
|
|
openapi_spec=collision_spec,
|
|
client=client,
|
|
name="New Server",
|
|
)
|
|
|
|
# Get tools from both servers
|
|
async with Client(legacy_server) as legacy_client:
|
|
legacy_tools = await legacy_client.list_tools()
|
|
|
|
async with Client(new_server) as new_client:
|
|
new_tools = await new_client.list_tools()
|
|
|
|
# Should have same number of tools
|
|
assert len(legacy_tools) == len(new_tools)
|
|
assert len(legacy_tools) == 1
|
|
|
|
# Get the single tool from each
|
|
legacy_tool = legacy_tools[0]
|
|
new_tool = new_tools[0]
|
|
|
|
# Input schemas should be identical
|
|
legacy_schema = legacy_tool.inputSchema
|
|
new_schema = new_tool.inputSchema
|
|
|
|
# Both should have collision-resolved parameters
|
|
legacy_props = legacy_schema.get("properties", {})
|
|
new_props = new_schema.get("properties", {})
|
|
|
|
# Should have: id__path (path param), id (body param), name (body param)
|
|
expected_props = {"id__path", "id", "name"}
|
|
assert set(legacy_props.keys()) == expected_props
|
|
assert set(new_props.keys()) == expected_props
|
|
|
|
# Required should include path param and required body params
|
|
legacy_required = set(legacy_schema.get("required", []))
|
|
new_required = set(new_schema.get("required", []))
|
|
assert legacy_required == new_required
|
|
assert "id__path" in legacy_required
|
|
assert "name" in legacy_required
|
|
|
|
# Path parameter should have integer type
|
|
assert legacy_props["id__path"]["type"] == "integer"
|
|
assert new_props["id__path"]["type"] == "integer"
|
|
|
|
# Body parameters should match
|
|
assert legacy_props["id"]["type"] == "integer"
|
|
assert new_props["id"]["type"] == "integer"
|
|
assert legacy_props["name"]["type"] == "string"
|
|
assert new_props["name"]["type"] == "string"
|
|
|
|
async def test_tool_execution_parameter_mapping(self, collision_spec):
|
|
"""Test that tool execution with collisions works identically."""
|
|
# This test verifies that both implementations can execute the same arguments
|
|
# We can't easily test actual HTTP calls, but we can test argument validation
|
|
|
|
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
|
|
# Create both servers
|
|
legacy_server = LegacyFastMCPOpenAPI(
|
|
openapi_spec=collision_spec,
|
|
client=client,
|
|
name="Legacy Server",
|
|
)
|
|
new_server = FastMCPOpenAPI(
|
|
openapi_spec=collision_spec,
|
|
client=client,
|
|
name="New Server",
|
|
)
|
|
|
|
# Test arguments that should work with collision resolution
|
|
test_args = {
|
|
"id__path": 123, # Path parameter (suffixed)
|
|
"id": 456, # Body parameter (not suffixed)
|
|
"name": "John Doe", # Body parameter
|
|
}
|
|
|
|
async with Client(legacy_server) as legacy_client:
|
|
async with Client(new_server) as new_client:
|
|
# Both should accept the same arguments
|
|
# We'll test this by attempting to call the tools
|
|
# (they'll fail at HTTP level but should pass argument validation)
|
|
|
|
legacy_tools = await legacy_client.list_tools()
|
|
new_tools = await new_client.list_tools()
|
|
|
|
legacy_tool_name = legacy_tools[0].name
|
|
new_tool_name = new_tools[0].name
|
|
|
|
# Names should be identical
|
|
assert legacy_tool_name == new_tool_name
|
|
|
|
# Both should fail at the HTTP request level (not argument validation)
|
|
# This confirms the argument mapping works identically
|
|
with pytest.raises(Exception) as legacy_exc:
|
|
await legacy_client.call_tool(legacy_tool_name, test_args)
|
|
|
|
with pytest.raises(Exception) as new_exc:
|
|
await new_client.call_tool(new_tool_name, test_args)
|
|
|
|
# Both should fail with similar error types (HTTP-related, not schema validation)
|
|
# The exact error might differ but shouldn't be schema validation errors
|
|
legacy_error = str(legacy_exc.value)
|
|
new_error = str(new_exc.value)
|
|
|
|
# Neither should fail due to schema validation
|
|
assert "schema" not in legacy_error.lower()
|
|
assert "schema" not in new_error.lower()
|
|
assert "validation" not in legacy_error.lower()
|
|
assert "validation" not in new_error.lower()
|
|
|
|
async def test_optional_parameter_handling(self, simple_spec):
|
|
"""Test that optional parameters are handled identically."""
|
|
async with httpx.AsyncClient(base_url="https://api.example.com") as client:
|
|
# Create both servers
|
|
legacy_server = LegacyFastMCPOpenAPI(
|
|
openapi_spec=simple_spec,
|
|
client=client,
|
|
name="Legacy Server",
|
|
)
|
|
new_server = FastMCPOpenAPI(
|
|
openapi_spec=simple_spec,
|
|
client=client,
|
|
name="New Server",
|
|
)
|
|
|
|
# Test with optional parameter omitted (should be None/null)
|
|
test_args_minimal = {"id": 123}
|
|
|
|
# Test with optional parameter included
|
|
test_args_full = {"id": 123, "include_details": True}
|
|
|
|
async with Client(legacy_server) as legacy_client:
|
|
async with Client(new_server) as new_client:
|
|
legacy_tools = await legacy_client.list_tools()
|
|
await new_client.list_tools()
|
|
|
|
tool_name = legacy_tools[0].name
|
|
|
|
# Both should handle minimal args the same way
|
|
with pytest.raises(Exception) as legacy_exc_min:
|
|
await legacy_client.call_tool(tool_name, test_args_minimal)
|
|
|
|
with pytest.raises(Exception) as new_exc_min:
|
|
await new_client.call_tool(tool_name, test_args_minimal)
|
|
|
|
# Both should handle full args the same way
|
|
with pytest.raises(Exception) as legacy_exc_full:
|
|
await legacy_client.call_tool(tool_name, test_args_full)
|
|
|
|
with pytest.raises(Exception) as new_exc_full:
|
|
await new_client.call_tool(tool_name, test_args_full)
|
|
|
|
# All should fail at HTTP level, not schema validation
|
|
for exc in [
|
|
legacy_exc_min,
|
|
new_exc_min,
|
|
legacy_exc_full,
|
|
new_exc_full,
|
|
]:
|
|
error_msg = str(exc.value).lower()
|
|
assert "schema" not in error_msg
|
|
assert "validation" not in error_msg
|