mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-10-16 18:03:19 +02:00
59d158a1ebWhen two routes produce the same component name, `OpenAPIProvider` appends a counter, so the second `list_items` becomes `list_items_2`. It never checked whether that name was already taken, and it never recorded the name it handed out. If another operation is itself named `list_items_2`, both components get the same name and the later one silently replaces the earlier one. This happens whether the suffixed name comes before or after the duplicates.
Suffixing now skips names that are already taken and records the name it assigns, so every route stays reachable. Names that didn't collide are unchanged.
```python
paths = {
"/v1/items": {"get": {"operationId": "list_items__v1", ...}},
"/v2/items": {"get": {"operationId": "list_items__v2", ...}},
"/archived-items": {"get": {"operationId": "list_items_2", ...}},
}
# before: ['list_items', 'list_items_2'] (one route lost)
# after: ['list_items', 'list_items_2', 'list_items_2_2']
```
Fixes #5322.
Co-authored-by: asemabdallah <asasem547@gmail.com>
446 lines
18 KiB
Python
446 lines
18 KiB
Python
"""Unit tests for OpenAPIProvider."""
|
|
|
|
import httpx2
|
|
import pytest
|
|
|
|
from fastmcp import FastMCP
|
|
from fastmcp.client import Client
|
|
from fastmcp.server.providers.openapi import OpenAPIProvider
|
|
from fastmcp.server.providers.openapi.provider import DEFAULT_TIMEOUT
|
|
from fastmcp.server.providers.openapi.routing import MCPType, RouteMap
|
|
|
|
|
|
class TestOpenAPIProviderServerVariables:
|
|
"""Test that OpenAPIProvider resolves OpenAPI 3.x server variables."""
|
|
|
|
def test_server_variables_substituted_with_defaults(self):
|
|
spec = {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Test API", "version": "1.0.0"},
|
|
"servers": [
|
|
{
|
|
"url": "https://{region}.api.example.com/v1",
|
|
"variables": {
|
|
"region": {
|
|
"default": "us",
|
|
"enum": ["us", "eu", "apac"],
|
|
}
|
|
},
|
|
}
|
|
],
|
|
"paths": {},
|
|
}
|
|
client = OpenAPIProvider._create_default_client(spec)
|
|
assert str(client.base_url) == "https://us.api.example.com/v1/"
|
|
|
|
def test_multiple_server_variables_substituted(self):
|
|
spec = {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Test API", "version": "1.0.0"},
|
|
"servers": [
|
|
{
|
|
"url": "{scheme}://{host}/v1",
|
|
"variables": {
|
|
"scheme": {"default": "https"},
|
|
"host": {"default": "api.example.com"},
|
|
},
|
|
}
|
|
],
|
|
"paths": {},
|
|
}
|
|
client = OpenAPIProvider._create_default_client(spec)
|
|
assert str(client.base_url) == "https://api.example.com/v1/"
|
|
|
|
def test_static_server_url_unaffected(self):
|
|
spec = {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Test API", "version": "1.0.0"},
|
|
"servers": [{"url": "https://api.example.com"}],
|
|
"paths": {},
|
|
}
|
|
client = OpenAPIProvider._create_default_client(spec)
|
|
assert str(client.base_url) == "https://api.example.com"
|
|
|
|
|
|
class TestOpenAPIProviderBasicFunctionality:
|
|
"""Test basic OpenAPIProvider 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_provider_initialization(self, simple_openapi_spec):
|
|
"""Test provider initialization with OpenAPI spec."""
|
|
client = httpx2.AsyncClient(base_url="https://api.example.com")
|
|
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec, client=client)
|
|
|
|
# Should have initialized RequestDirector successfully
|
|
assert hasattr(provider, "_director")
|
|
assert hasattr(provider, "_spec")
|
|
|
|
def test_server_with_provider(self, simple_openapi_spec):
|
|
"""Test server initialization with OpenAPIProvider."""
|
|
client = httpx2.AsyncClient(base_url="https://api.example.com")
|
|
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec, client=client)
|
|
|
|
mcp = FastMCP("Test Server")
|
|
mcp.add_provider(provider)
|
|
|
|
assert mcp.name == "Test Server"
|
|
|
|
async def test_provider_creates_tools_from_spec(self, simple_openapi_spec):
|
|
"""Test that provider creates tools from OpenAPI spec."""
|
|
async with httpx2.AsyncClient(base_url="https://api.example.com") as client:
|
|
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec, client=client)
|
|
|
|
mcp = FastMCP("Test Server")
|
|
mcp.add_provider(provider)
|
|
|
|
async with Client(mcp) 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_provider_tool_execution(self, simple_openapi_spec):
|
|
"""Test tool execution uses RequestDirector."""
|
|
mock_client = httpx2.AsyncClient()
|
|
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec, client=mock_client)
|
|
|
|
mcp = FastMCP("Test Server")
|
|
mcp.add_provider(provider)
|
|
|
|
async with Client(mcp) as mcp_client:
|
|
tools = await mcp_client.list_tools()
|
|
|
|
# Should have tools using RequestDirector
|
|
assert len(tools) == 2
|
|
|
|
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_provider_creates_default_client_from_spec(self, simple_openapi_spec):
|
|
"""Test that omitting client creates one from the spec's servers URL."""
|
|
provider = OpenAPIProvider(openapi_spec=simple_openapi_spec)
|
|
assert str(provider._client.base_url).rstrip("/") == "https://api.example.com"
|
|
assert provider._client.timeout == httpx2.Timeout(DEFAULT_TIMEOUT)
|
|
|
|
def test_provider_default_client_requires_servers(self):
|
|
"""Test that omitting client without servers in spec raises."""
|
|
spec = {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "No Servers", "version": "1.0.0"},
|
|
"paths": {},
|
|
}
|
|
with pytest.raises(ValueError, match="No server URL"):
|
|
OpenAPIProvider(openapi_spec=spec)
|
|
|
|
def test_provider_with_empty_spec(self):
|
|
"""Test provider with minimal OpenAPI spec."""
|
|
minimal_spec = {
|
|
"openapi": "3.0.0",
|
|
"info": {"title": "Empty API", "version": "1.0.0"},
|
|
"paths": {},
|
|
}
|
|
|
|
client = httpx2.AsyncClient(base_url="https://api.example.com")
|
|
provider = OpenAPIProvider(openapi_spec=minimal_spec, client=client)
|
|
|
|
# Should handle empty paths gracefully
|
|
assert hasattr(provider, "_director")
|
|
assert hasattr(provider, "_spec")
|
|
|
|
async def test_clean_schema_output_no_unused_defs(self):
|
|
"""Test that unused schema definitions are removed from tool schemas."""
|
|
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": {
|
|
"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 httpx2.AsyncClient(base_url="https://api.example.com") as client:
|
|
provider = OpenAPIProvider(
|
|
openapi_spec=spec_with_unused_defs, client=client
|
|
)
|
|
mcp = FastMCP("Test Server")
|
|
mcp.add_provider(provider)
|
|
|
|
async with Client(mcp) as mcp_client:
|
|
tools = await mcp_client.list_tools()
|
|
|
|
assert len(tools) == 1
|
|
tool = tools[0]
|
|
|
|
assert tool.name == "create_user"
|
|
|
|
expected_input_schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"name": {"type": "string", "title": "Name"},
|
|
"active": {"type": "boolean", "title": "Active"},
|
|
},
|
|
"required": ["name", "active"],
|
|
}
|
|
assert tool.input_schema == expected_input_schema
|
|
|
|
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.output_schema == expected_output_schema
|
|
|
|
|
|
def names_spec(*routes: tuple[str, str]) -> dict:
|
|
return {
|
|
"openapi": "3.1.0",
|
|
"info": {"title": "Names", "version": "1.0.0"},
|
|
"paths": {
|
|
path: {
|
|
"get": {
|
|
"operationId": operation_id,
|
|
"responses": {"200": {"description": "OK"}},
|
|
}
|
|
}
|
|
for path, operation_id in routes
|
|
},
|
|
}
|
|
|
|
|
|
class TestOpenAPIProviderComponentNames:
|
|
"""Generated component names stay unique so no route is silently dropped."""
|
|
|
|
async def test_duplicate_names_get_numbered_suffixes(self):
|
|
spec = names_spec(
|
|
("/a", "list_items__a"), ("/b", "list_items__b"), ("/c", "list_items__c")
|
|
)
|
|
server = FastMCP.from_openapi(
|
|
spec, client=httpx2.AsyncClient(base_url="http://test")
|
|
)
|
|
|
|
async with Client(server) as client:
|
|
tools = await client.list_tools()
|
|
|
|
assert sorted(tool.name for tool in tools) == [
|
|
"list_items",
|
|
"list_items_2",
|
|
"list_items_3",
|
|
]
|
|
|
|
@pytest.mark.parametrize(
|
|
"routes",
|
|
[
|
|
[("/a", "list_items__a"), ("/b", "list_items__b"), ("/c", "list_items_2")],
|
|
[("/c", "list_items_2"), ("/a", "list_items__a"), ("/b", "list_items__b")],
|
|
],
|
|
)
|
|
async def test_suffix_does_not_reuse_an_existing_name(
|
|
self, routes: list[tuple[str, str]]
|
|
):
|
|
requested_paths: list[str] = []
|
|
|
|
def handler(request: httpx2.Request) -> httpx2.Response:
|
|
requested_paths.append(request.url.path)
|
|
return httpx2.Response(200, json={})
|
|
|
|
api = httpx2.AsyncClient(
|
|
base_url="http://test", transport=httpx2.MockTransport(handler)
|
|
)
|
|
server = FastMCP.from_openapi(names_spec(*routes), client=api)
|
|
|
|
async with Client(server) as client:
|
|
tools = await client.list_tools()
|
|
for tool in tools:
|
|
await client.call_tool(tool.name, {})
|
|
|
|
assert len(tools) == 3
|
|
assert len({tool.name for tool in tools}) == 3
|
|
assert sorted(requested_paths) == ["/a", "/b", "/c"]
|
|
|
|
async def test_resource_suffix_does_not_reuse_an_existing_name(self):
|
|
spec = names_spec(
|
|
("/a", "list_items__a"), ("/b", "list_items__b"), ("/c", "list_items_2")
|
|
)
|
|
server = FastMCP.from_openapi(
|
|
spec,
|
|
client=httpx2.AsyncClient(base_url="http://test"),
|
|
route_maps=[RouteMap(mcp_type=MCPType.RESOURCE)],
|
|
)
|
|
|
|
async with Client(server) as client:
|
|
resources = await client.list_resources()
|
|
|
|
assert len({str(resource.uri) for resource in resources}) == 3
|