Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions fastapi_mcp/server.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import hashlib
import json
import httpx
from typing import Dict, Optional, Any, List, Union, Literal, Sequence
Expand All @@ -24,6 +25,8 @@ class FastApiMCP:
Create an MCP server from a FastAPI app.
"""

_MAX_COMBINED_TOOL_NAME_LENGTH = 60

def __init__(
self,
fastapi: Annotated[
Expand Down Expand Up @@ -140,6 +143,7 @@ def setup_server(self) -> None:

# Filter tools based on operation IDs and tags
self.tools = self._filter_tools(all_tools, openapi_schema)
self.tools = self._shorten_tool_names_for_client_limits(self.tools)

mcp_server: Server = Server(self.name, self.description)

Expand Down Expand Up @@ -185,6 +189,53 @@ async def handle_call_tool(

self.server = mcp_server

@classmethod
def _build_short_tool_name(cls, tool_name: str, max_length: int) -> str:
if len(tool_name) <= max_length:
return tool_name

name_hash = hashlib.sha1(tool_name.encode("utf-8")).hexdigest()[:8]
suffix = f"-{name_hash}"

if max_length <= len(suffix):
return name_hash[:max_length]

return f"{tool_name[: max_length - len(suffix)]}{suffix}"

def _shorten_tool_names_for_client_limits(self, tools: List[types.Tool]) -> List[types.Tool]:
max_tool_name_length = self._MAX_COMBINED_TOOL_NAME_LENGTH - len(self.name)

if max_tool_name_length < 1:
logger.warning(
"Server name '%s' is too long to satisfy the %s-character combined client tool limit",
self.name,
self._MAX_COMBINED_TOOL_NAME_LENGTH,
)
return tools

shortened_tools: List[types.Tool] = []
shortened_operation_map: Dict[str, Dict[str, Any]] = {}
used_names: set[str] = set()

for tool in tools:
original_name = tool.name
shortened_name = self._build_short_tool_name(original_name, max_tool_name_length)

if shortened_name in used_names:
collision_index = 1
while shortened_name in used_names:
collision_suffix = f"-{collision_index}"
available_length = max(1, max_tool_name_length - len(collision_suffix))
shortened_name = f"{self._build_short_tool_name(original_name, available_length)}{collision_suffix}"
collision_index += 1

used_names.add(shortened_name)
shortened_tools.append(tool.model_copy(update={"name": shortened_name}))
shortened_operation_map[shortened_name] = self.operation_map[original_name]

self.operation_map = shortened_operation_map
return shortened_tools

def _register_mcp_connection_endpoint_sse(
self,
router: FastAPI | APIRouter,
Expand Down
24 changes: 24 additions & 0 deletions tests/test_configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -581,3 +581,27 @@ async def empty_tags():
exclude_tags_mcp = FastApiMCP(app, exclude_tags=["items"])
assert len(exclude_tags_mcp.tools) == 1
assert {tool.name for tool in exclude_tags_mcp.tools} == {"empty_tags"}


def test_tool_names_are_shortened_to_fit_client_limits():
"""Test that tool names are shortened when combined server + tool names exceed common client limits."""
app = FastAPI(title="tool-shortening-test")

long_operation_id = "operation_name_that_is_far_too_long_for_cursor_tool_limits"
custom_server_name = "custom-server-name-with-limit"

@app.get("/items/", operation_id=long_operation_id)
async def list_items():
return [{"id": 1}]

mcp_server = FastApiMCP(app, name=custom_server_name)

assert len(mcp_server.tools) == 1

tool_name = mcp_server.tools[0].name
max_tool_name_length = 60 - len(custom_server_name)

assert tool_name != long_operation_id
assert len(tool_name) <= max_tool_name_length
assert tool_name in mcp_server.operation_map
assert long_operation_id not in mcp_server.operation_map
41 changes: 41 additions & 0 deletions tests/test_mcp_execute_api_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -190,3 +190,44 @@ async def test_execute_api_tool_with_non_ascii_chars(simple_fastapi_app: FastAPI
params={},
headers={}
)


@pytest.mark.asyncio
async def test_execute_api_tool_with_shortened_tool_name():
"""Test executing a tool after its name is shortened to satisfy client limits."""
app = FastAPI(title="tool-shortening-test")
custom_server_name = "custom-server-name-with-limit"
long_operation_id = "operation_name_that_is_far_too_long_for_cursor_tool_limits"

@app.get("/items/{item_id}", operation_id=long_operation_id)
async def get_item(item_id: int):
return {"id": item_id}

mcp = FastApiMCP(app, name=custom_server_name)

tool_name = mcp.tools[0].name
assert tool_name != long_operation_id

mock_response = MagicMock()
mock_response.json.return_value = {"id": 1}
mock_response.status_code = 200
mock_response.text = '{"id": 1}'

mock_client = AsyncMock()
mock_client.get.return_value = mock_response

with patch.object(mcp, "_http_client", mock_client):
result = await mcp._execute_api_tool(
client=mock_client,
tool_name=tool_name,
arguments={"item_id": 1},
operation_map=mcp.operation_map,
)

assert len(result) == 1
assert isinstance(result[0], TextContent)
mock_client.get.assert_called_once_with(
"/items/1",
params={},
headers={},
)