Condensed OpenAPI

[shane-davidson]

Some APIs schemas are massive and the model/client doesn't like loading so many tools.

This class wraps an entire API schema inside the following objects:

list_objects -> list of what would be the old openAPI tools
list_object_schema -> list tool/object schema
call_object -> calls the old tool by name

I still haven't finished the cleaning up the returned data for list_objects/list_objects_schema

Happy for you guys to do what you want with this idea :)

class FastMCPCompressedOpenAPI(FastMCPOpenAPI):
    """
    Compressed FastMCP OpenAPI implementation that maintains full compatibility
    with FastMCPOpenAPI but exposes only 3 tools instead of individual endpoint tools.

    This class inherits from FastMCPOpenAPI and overrides the component creation
    to provide a compressed interface while preserving all the original functionality.
    """

    def __init__(self, *args, **kwargs):
        """
        Initialize the compressed OpenAPI server.

        All parameters are passed through to FastMCPOpenAPI, but individual
        endpoint tools are replaced with 3 compressed tools.
        """
        # Extract openapi_spec and ensure it's passed as keyword argument
        openapi_spec = kwargs.get("openapi_spec")
        if openapi_spec is None and args:
            # Assume first positional argument is openapi_spec if not given as kwarg
            openapi_spec = args[0]
            kwargs["openapi_spec"] = openapi_spec

        # Ensure openapi_spec is not None
        if openapi_spec is None:
            openapi_spec = {}

        # Call parent constructor to set up all the infrastructure
        super().__init__(*args, **kwargs)

        # Store the openapi_spec for later use
        self._openapi_spec = openapi_spec

        # Copy the tools registered by the parent
        self._original_tools = list(self._tool_manager._tools.values())

        # Clear the individual tools that were created by the parent
        self._tool_manager._tools.clear()

        # Register the 3 compressed tools
        self._register_compressed_tools()

    def _tool_to_dict(self, tool: Tool) -> dict:
        """Convert an OpenAPITool to a dictionary."""
        # Try to get method/path from the tool, else from the underlying route
        method = getattr(tool, "method", None)
        path = getattr(tool, "path", None)
        operation_id = getattr(tool, "operation_id", None)
        summary = getattr(tool, "summary", None)
        # If not set, try to get from route
        route = getattr(tool, "route", None)
        if route:
            method = method or getattr(route, "method", None)
            path = path or getattr(route, "path", None)
            operation_id = operation_id or getattr(route, "operation_id", None)
            summary = summary or getattr(route, "summary", None)
        return {
            "name": tool.name,
            "method": method,
            "path": path,
            "summary": summary,
            "description": getattr(tool, "description", None),
            "tags": list(getattr(tool, "tags", [])),
            "operation_id": operation_id,
        }

    def _register_compressed_tools(self):
        """Register the 3 compressed tools that provide access to all endpoints."""

        # Use the tools copied from the parent
        tool_objs = self._original_tools

        async def list_objects() -> dict:
            """List all available API endpoints."""
            return {
                "endpoints": [self._tool_to_dict(tool) for tool in tool_objs],
                "total": len(tool_objs),
            }

        async def list_object_schema(object_name: str) -> dict:
            """Get the schema for a specific endpoint."""
            for tool in tool_objs:
                if tool.name == object_name:
                    return self._tool_to_dict(tool)
            available_objects = [tool.name for tool in tool_objs]
            raise ValueError(
                f"Object '{object_name}' not found. Available objects: {available_objects}"
            )

        async def call_object(object_name: str, params: dict) -> dict:
            """Execute a specific API endpoint with the given parameters."""
            for tool in tool_objs:
                if tool.name == object_name:
                    result = await tool.run(params)
                    if result and len(result) > 0:
                        text_content = result[0]
                        if (
                            hasattr(text_content, "text")
                            and hasattr(text_content, "type")
                            and text_content.type == "text"
                        ):
                            return {"result": text_content.text}
                        return {"result": str(text_content)}
                    return {"result": ""}
            available_objects = [tool.name for tool in tool_objs]
            raise ValueError(
                f"Object '{object_name}' not found. Available objects: {available_objects}"
            )

        # Register the compressed tools
        self.tool(list_objects)
        self.tool(list_object_schema)
        self.tool(call_object)

[jlowin]

In general, I think turning massive schemas into MCP servers is a smell - there's no way for an LLM to perform well in the face of such choice, a tailored MCP is going to be much better. The conversion functionality is more of a convenience rather than a panacea. That said, I think experimenting with different approaches to exposing that utility is interesting. I don't think this rises to the level of a pattern we'd want to bless as universally applicable but could be an interesting contrib/ module?