API Reference¶

This page provides detailed API documentation for the MCP Compressor package.

Main Module¶

Main entry point for the MCP Compressor CLI.

This module provides the CLI interface for running the MCP Compressor proxy server, which wraps existing MCP servers and compresses their tool descriptions to reduce token consumption.

`clear_oauth(all_tokens=False)` ¶

Clear stored OAuth tokens for all servers.

Removes cached OAuth tokens so the next connection will re-authenticate. By default the encryption key is preserved so new tokens are stored under the same key. Pass --all to also remove the key itself.

Source code in mcp_compressor/main.py

@clear_oauth_app.callback(invoke_without_command=True)
def clear_oauth(
    all_tokens: Annotated[
        bool,
        typer.Option("--all", help="Also delete the encryption key, forcing full re-authentication next run."),
    ] = False,
) -> None:
    """Clear stored OAuth tokens for all servers.

    Removes cached OAuth tokens so the next connection will re-authenticate.
    By default the encryption key is preserved so new tokens are stored under
    the same key.  Pass --all to also remove the key itself.
    """
    token_dir = _OAUTH_CONFIG_DIR / "oauth-tokens"
    key_file = _OAUTH_CONFIG_DIR / ".key"
    removed: list[str] = []

    if token_dir.exists():
        shutil.rmtree(token_dir)
        removed.append(str(token_dir))

    if all_tokens and key_file.exists():
        key_file.unlink()
        removed.append(str(key_file))

    if removed:
        print("Removed:")
        for path in removed:
            print(f"  {path}")
        # Also clear from keyring if present
        if all_tokens:
            with contextlib.suppress(Exception):
                keyring.delete_password("mcp-compressor", "encryption-key")
                print("  keyring entry: mcp-compressor / encryption-key")
        print("OAuth credentials cleared. You will be prompted to authenticate on next connection.")
    else:
        print("No stored OAuth credentials found.")

`entrypoint()` ¶

Main entrypoint for the mcp-compressor CLI.

Handles the 'clear-oauth' subcommand manually before delegating to the main Typer app, so that 'mcp-compressor ' works without a subcommand.

Source code in mcp_compressor/main.py

def entrypoint() -> None:
    """Main entrypoint for the mcp-compressor CLI.

    Handles the 'clear-oauth' subcommand manually before delegating to the
    main Typer app, so that 'mcp-compressor <url>' works without a subcommand.
    """
    if len(sys.argv) > 1 and sys.argv[1] == "clear-oauth":
        sys.argv = [sys.argv[0], *sys.argv[2:]]
        clear_oauth_app()
    else:
        app()

`main(command_or_url_list, cwd=None, env_list=None, header_list=None, timeout=10.0, compression_level=CompressionLevel.MEDIUM, server_name=None, log_level=LogLevel.WARNING, toonify=False, cli_mode=False, cli_port=None)` ¶

Run the MCP Compressor proxy server.

This is the main entry point for the CLI application. It connects to an MCP server (via stdio, HTTP, or SSE) and wraps it with a compressed tool interface.

Source code in mcp_compressor/main.py

@app.command()
def main(
    command_or_url_list: Annotated[
        list[str],
        typer.Argument(
            ...,
            metavar="COMMAND_OR_URL",
            help=(
                "The URL of the MCP server to connect to for streamable HTTP or SSE servers, or the command and "
                "arguments to run for stdio servers. Example: uvx mcp-server-fetch"
            ),
        ),
    ],
    cwd: Annotated[
        str | None,
        typer.Option(
            ...,
            "--cwd",
            help="The working directory to use when running stdio MCP servers.",
        ),
    ] = None,
    env_list: Annotated[
        list[str] | None,
        typer.Option(
            ...,
            "--env",
            "-e",
            help=(
                "Environment variables to set when running stdio MCP servers, in the form VAR_NAME=VALUE. Can be used "
                "multiple times. Supports environment variable expansion with ${VAR_NAME} syntax."
            ),
        ),
    ] = None,
    header_list: Annotated[
        list[str] | None,
        typer.Option(
            ...,
            "--header",
            "-H",
            help=(
                "Headers to use for remote (HTTP/SSE) MCP server connections, in the form Header-Name=Header-Value. "
                "Can be use multiple times. Supports environment variable expansion with ${VAR_NAME} syntax."
            ),
        ),
    ] = None,
    timeout: Annotated[
        float,
        typer.Option(
            ...,
            "--timeout",
            "-t",
            help="The timeout in seconds for connecting to the MCP server and making requests.",
        ),
    ] = 10.0,
    compression_level: Annotated[
        CompressionLevel,
        typer.Option(
            ...,
            "--compression-level",
            "-c",
            help=("The level of compression to apply to tool the tools descriptions of the wrapped MCP server."),
            case_sensitive=False,
        ),
    ] = CompressionLevel.MEDIUM,
    server_name: Annotated[
        str | None,
        typer.Option(
            ...,
            "--server-name",
            "-n",
            help=(
                "Optional custom name to prefix the wrapper tool names (get_tool_schema, invoke_tool, list_tools). "
                "The name will be sanitized to conform to MCP tool name specifications (only A-Z, a-z, 0-9, _, -, .)."
            ),
        ),
    ] = None,
    log_level: Annotated[
        LogLevel,
        typer.Option(
            ...,
            "--log-level",
            "-l",
            help=(
                "The logging level. Used for both the MCP Compressor server and the underlying MCP server if it is a "
                "stdio server."
            ),
            case_sensitive=False,
        ),
    ] = LogLevel.WARNING,
    toonify: Annotated[
        bool,
        typer.Option(..., "--toonify", help="Convert JSON tool responses to TOON format automatically."),
    ] = False,
    cli_mode: Annotated[
        bool,
        typer.Option(
            ...,
            "--cli-mode",
            help=(
                "Start in CLI mode: expose a single help MCP tool, start a local HTTP bridge, "
                "and generate a shell script for interacting with the wrapped server via CLI. "
                "--toonify is automatically enabled in this mode."
            ),
        ),
    ] = False,
    cli_port: Annotated[
        int | None,
        typer.Option(
            ...,
            "--cli-port",
            help="Port for the local CLI bridge HTTP server (default: random free port).",
        ),
    ] = None,
):
    """Run the MCP Compressor proxy server.

    This is the main entry point for the CLI application. It connects to an MCP server
    (via stdio, HTTP, or SSE) and wraps it with a compressed tool interface.
    """
    logger.remove()
    logger.add(sys.stderr, level=log_level.value.upper())
    setup_loguru_logging_intercept(modules=("fastmcp",))

    if threading.current_thread() is threading.main_thread():
        shutting_down = False

        def _handle_interrupt(signum: int, frame: object) -> None:
            nonlocal shutting_down
            if shutting_down:
                logger.debug("Ignoring additional interrupt signal during shutdown")
                return
            shutting_down = True
            logger.info("Server stopped")
            # Terminate child processes (stdio backend server) to avoid zombies
            with contextlib.suppress(Exception):
                current = psutil.Process()
                for child in current.children(recursive=True):
                    with contextlib.suppress(Exception):
                        child.terminate()
            # os._exit(0) bypasses daemon thread join hangs (both stdio stdin-read
            # threads and HTTP transport threads can block interpreter shutdown)
            os._exit(0)

        signal.signal(signal.SIGINT, _handle_interrupt)
        signal.signal(signal.SIGTERM, _handle_interrupt)

    asyncio.run(
        _async_main(
            command_or_url_list=command_or_url_list,
            cwd=cwd,
            env_list=env_list,
            header_list=header_list,
            timeout=timeout,
            compression_level=compression_level,
            server_name=server_name,
            log_level=log_level,
            toonify=toonify or cli_mode,
            cli_mode=cli_mode,
            cli_port=cli_port,
        )
    )

Tools Module¶

Tool compression middleware for MCP servers.

This module provides the CompressedTools middleware that wraps MCP server tools and compresses their descriptions to reduce token consumption while maintaining full functionality through a two-step tool invocation pattern.

`CompressedTools` ¶

Bases: Middleware

Middleware that compresses MCP tool descriptions to reduce token consumption.

This middleware wraps an MCP client and exposes its tools through a compressed interface. In normal mode it provides two or three public wrapper tools: - get_tool_schema: Retrieves the full schema for a specific tool - invoke_tool: Executes a tool with the provided arguments - list_tools: (optional) Lists all available tools with brief descriptions (only if compression level is MAX)

It also registers a resource (compressor://uncompressed-tools) that exposes the upstream server's original list_tools payload in machine-readable JSON form.

In CLI mode it provides a single help tool (_help) that lists all CLI subcommands. The compression level determines how much information is included in the get_tool_schema tool description.

Source code in mcp_compressor/tools.py

class CompressedTools(Middleware):
    """Middleware that compresses MCP tool descriptions to reduce token consumption.

    This middleware wraps an MCP client and exposes its tools through a compressed interface.
    In normal mode it provides two or three public wrapper tools:
    - get_tool_schema: Retrieves the full schema for a specific tool
    - invoke_tool: Executes a tool with the provided arguments
    - list_tools: (optional) Lists all available tools with brief descriptions (only if compression level is MAX)

    It also registers a resource (``compressor://uncompressed-tools``) that exposes the upstream server's
    original list_tools payload in machine-readable JSON form.

    In CLI mode it provides a single help tool (<server_name>_help) that lists all CLI subcommands.
    The compression level determines how much information is included in the get_tool_schema tool description.
    """

    def __init__(
        self,
        proxy_server: FastMCP,
        compression_level: CompressionLevel,
        server_name: str | None = None,
        toonify: bool = False,
        cli_mode: bool = False,
        cli_name: str | None = None,
    ) -> None:
        """Initialize the CompressedTools middleware.

        Args:
            proxy_server: The MCP proxy server instance.
            compression_level: The level of compression to apply to tool descriptions.
            server_name: Optional custom name prefix for tool names (will be sanitized and used as prefix).
            toonify: Whether to convert JSON text tool outputs to TOON format.
            cli_mode: Whether to run in CLI mode (exposes a single help tool instead of wrapper tools).
            cli_name: The CLI script name (used in CLI mode for help text).
        """
        self._proxy_server = proxy_server
        self._compression_level = compression_level
        self._tool_name_prefix = f"{server_name}_" if server_name else ""
        self._server_description = f"the {server_name} toolset" if server_name else "this toolset"
        self._toonify = toonify
        self._cli_mode = cli_mode
        self._cli_name = cli_name or (server_name or "mcp")
        self._help_tool_name = sanitize_tool_name(f"{server_name}_help" if server_name else "help")
        self._get_schema_tool_name = sanitize_tool_name(f"{self._tool_name_prefix}get_tool_schema")
        self._invoke_tool_name = sanitize_tool_name(f"{self._tool_name_prefix}invoke_tool")
        self._invoke_tool_alias_name = sanitize_tool_name("invoke_tool")
        self._list_tools_name = sanitize_tool_name(f"{self._tool_name_prefix}list_tools")
        self._uncompressed_tools_resource_uri = "compressor://uncompressed-tools"
        self._hidden_tool_names: set[str] = set()
        self._built_in_tool_names = {self._get_schema_tool_name, self._invoke_tool_name}
        if self._invoke_tool_alias_name != self._invoke_tool_name:
            self._built_in_tool_names.add(self._invoke_tool_alias_name)
            self._hidden_tool_names.add(self._invoke_tool_alias_name)
        if self._compression_level == CompressionLevel.MAX:
            self._built_in_tool_names.add(self._list_tools_name)
        self._all_tools: dict[str, Tool] | None = None

    async def list_tools(self) -> str:
        """List all available tools in {server_description}.

        Returns:
            A formatted string listing tool names and brief descriptions.
        """
        return await self._get_tool_descriptions(CompressionLevel.MEDIUM)

    async def get_tool_schema(self, tool_name: str) -> str:
        """Get the input schema for a specific tool from {server_description}.

        Available tools are:
        {tool_descriptions}

        Args:
            tool_name: The name of the tool to get the schema for.

        Returns:
            The input schema for the specified tool.

        Raises:
            ValueError: If the tool name is not found in the server.
        """
        tool = await self._get_backend_tool(tool_name)
        tool_description = self._format_tool_description(tool, CompressionLevel.LOW)
        return tool_description + "\n\n" + json.dumps(tool.parameters, indent=2)

    async def invoke_tool(self, tool_name: str, tool_input: dict[str, Any] | None = None, quiet: bool = False) -> Any:
        """Invoke a tool from {server_description}.

        Args:
            tool_name: The name of the tool to invoke.
            tool_input: The input to the tool. Schemas can be retrieved using the appropriate `get_tool_schema`
                function.
            quiet: If true, truncates large tool outputs for successful invocations. This is useful for reducing token
                consumption when the output is not needed. Full responses will always be returned for tool errors.

        Returns:
            The output from the tool.
        """
        tool = await self._get_backend_tool(tool_name)
        try:
            tool_result = await tool.run(tool_input or {})
        except ValidationError as exc:
            raise ToolError(await self._format_validation_error(tool_name, str(exc))) from exc
        except ToolError as exc:
            if self._is_validation_error_message(str(exc)):
                raise ToolError(await self._format_validation_error(tool_name, str(exc))) from exc
            raise
        if self._toonify:
            tool_result = self._toonify_tool_result(tool_result)
        if quiet:
            if len(tool_result.content) == 1 and isinstance(tool_result.content[0], TextContent):
                return_text = tool_result.content[0].text
                if len(return_text) < QUIET_MODE_THRESHOLD:
                    return tool_result
                preview_length = QUIET_MODE_THRESHOLD // 2
                return_text = (
                    return_text[:preview_length]
                    + "\n...\n(truncated due to quiet mode)\n...\n"
                    + return_text[-preview_length:]
                )
            else:
                return_text = f"Successfully executed tool '{tool.name}' without output."
            return ToolResult(content=[TextContent(type="text", text=return_text)])
        return tool_result

    async def list_uncompressed_tools(self) -> str:
        """Return the upstream server's original list_tools payload as JSON.

        This hidden helper is intended for MCP-aware clients that need the backend server's uncompressed tool
        inventory, including the same descriptions and schemas exposed by the upstream list_tools endpoint.

        Returns:
            A JSON array matching the upstream server's list_tools response.
        """
        tools = []
        for tool in (await self._get_backend_tools()).values():
            tool_data = tool.model_dump(mode="json")
            tools.append({
                "name": tool_data["name"],
                "title": tool_data["title"],
                "description": tool_data["description"],
                "inputSchema": tool_data["parameters"],
                "outputSchema": tool_data["output_schema"],
                "icons": tool_data["icons"],
                "annotations": tool_data["annotations"],
                "meta": tool_data["meta"],
                "execution": tool_data["execution"],
            })
        return json.dumps(tools, indent=2)

    async def on_call_tool(
        self,
        context: MiddlewareContext[CallToolRequestParams],
        call_next: CallNext[CallToolRequestParams, ToolResult],
    ) -> ToolResult:
        """Middleware to clean up tool invocation arguments to invoke_tool and route to the underlying tool.

        Args:
            context: The middleware context containing the call request.
            call_next: The next middleware or handler in the chain.

        Returns:
            The result from calling the underlying tool.
        """
        tool_args = context.message.arguments
        if not context.message.name.endswith("invoke_tool") or not tool_args:
            result = await call_next(context)
            if self._toonify and not self._is_built_in_tool_name(context.message.name):
                return self._toonify_tool_result(result)
            return result

        if "tool_input" not in tool_args or tool_args["tool_input"] is None:
            tool_input = {k: v for k, v in tool_args.items() if k not in ["tool_name", "quiet"]}
        else:
            tool_input = tool_args["tool_input"]
        return await self.invoke_tool(
            tool_name=tool_args["tool_name"],
            tool_input=tool_input,
            quiet=tool_args.get("quiet", False),
        )

    async def on_list_tools(
        self, context: MiddlewareContext[ListToolsRequest], call_next: CallNext[ListToolsRequest, Sequence[Tool]]
    ) -> Sequence[Tool]:
        """Middleware to inject compressed tool descriptions and suppress backend tools.

        In normal mode, updates get_tool_schema's description with the tool list.
        In CLI mode, updates the help tool's description with the full CLI help text.

        Returns:
            The sequence of built-in wrapper tools with updated descriptions.
        """
        built_in_tools = [
            tool for tool in (await self._get_built_in_tools()).values() if not self._is_hidden_tool_name(tool.name)
        ]
        if self._cli_mode:
            description = await self._build_cli_description()
            for tool in built_in_tools:
                tool.description = description
            return built_in_tools
        prepared_tools = []
        for tool in built_in_tools:
            logger.info(f"Preparing tool: {tool.name}")
            prepared_tools.append(tool)
            tool.description = cast(str, tool.description).format(
                tool_descriptions=await self._get_tool_descriptions(self._compression_level),
                server_description=self._server_description,
            )
        return prepared_tools

    async def configure_server(self) -> None:
        """Configure an MCP server with compressed tool wrappers.

        In normal mode, registers get_tool_schema, invoke_tool, and optionally list_tools.
        In CLI mode, registers a single <server_name>_help tool.
        """
        if self._cli_mode:

            async def help_tool() -> str:
                return await self._build_cli_description()

            help_tool.__doc__ = f"Get help for the '{self._cli_name}' CLI. Lists all available subcommands."
            self._proxy_server.tool(name=self._help_tool_name)(help_tool)
        else:
            # Create tool names with optional server name prefix
            self._proxy_server.tool(name=self._get_schema_tool_name)(self.get_tool_schema)
            self._proxy_server.tool(name=self._invoke_tool_name)(self.invoke_tool)
            if self._invoke_tool_alias_name != self._invoke_tool_name:
                self._proxy_server.tool(name=self._invoke_tool_alias_name)(self.invoke_tool)
            self._proxy_server.resource(
                uri=self._uncompressed_tools_resource_uri,
                description="The upstream server's original uncompressed tool list as JSON.",
                mime_type="application/json",
            )(self.list_uncompressed_tools)
            if self._compression_level == CompressionLevel.MAX:
                self._proxy_server.tool(name=self._list_tools_name)(self.list_tools)
        self._proxy_server.add_middleware(self)
        self._all_tools = None  # Reset cached tools, if any

    async def get_compression_stats(self) -> dict[str, Any]:
        """Get statistics about the compression of tool descriptions.

        Computes the original backend schema size vs the compressed proxy tool size
        for all compression levels. Works identically in both normal and CLI mode —
        the only difference is which tools _get_built_in_tools() returns.

        Returns:
            A dictionary containing statistics about the original and compressed tool description sizes.
        """
        backend_tools = await self._get_backend_tools()
        original_tool_count = len(backend_tools)
        original_schema_size = sum(
            len(json.dumps(tool.parameters)) + len(json.dumps(tool.output_schema)) + len(tool.description or "")
            for tool in backend_tools.values()
        )
        # Low/Medium/High/Max: always measured from the backend tools compressed at each level.
        # This is what a non-CLI agent would see in the get_tool_schema description,
        # and it gives meaningful differentiation between levels in both modes.
        compressed_tool_count = len(backend_tools)
        compressed_schema_sizes: dict[CompressionLevel | str, int] = {}

        for compression_level in [
            CompressionLevel.LOW,
            CompressionLevel.MEDIUM,
            CompressionLevel.HIGH,
            CompressionLevel.MAX,
        ]:
            compressed_schema_sizes[compression_level] = sum(
                len(self._format_tool_description(tool, compression_level)) for tool in backend_tools.values()
            )

        # "cli" key: the help tool description — what the agent sees in CLI mode.
        compressed_schema_sizes["cli"] = len(await self._build_cli_description())
        return {
            "original_tool_count": original_tool_count,
            "compressed_tool_count": compressed_tool_count,
            "original_schema_size": original_schema_size,
            "compressed_schema_sizes": compressed_schema_sizes,
        }

    async def _build_cli_description(self) -> str:
        """Build the full help description for CLI mode — same content as the CLI --help output."""
        _, on_path = find_script_dir()
        tools = list((await self._get_backend_tools()).values())
        return build_help_tool_description(self._cli_name, self._server_description, tools, on_path=on_path)

    async def _get_tool_descriptions(self, compression_level: CompressionLevel) -> str:
        """Generate a formatted string of tool descriptions at the specified compression level.

        Args:
            compression_level: The compression level to use for formatting.

        Returns:
            A newline-separated string of formatted tool descriptions.
        """
        if compression_level == CompressionLevel.MAX:
            return ""
        tool_descriptions = []
        for tool in (await self._get_backend_tools()).values():
            tool_descriptions.append(self._format_tool_description(tool, compression_level))
        return "\n".join(tool_descriptions)

    def _format_tool_description(self, tool: Tool, compression_level: CompressionLevel) -> str:
        """Format a single tool's description based on the compression level.

        Args:
            tool: The tool to format.
            compression_level: The compression level determining how much detail to include.

        Returns:
            A formatted string representation of the tool in the format:
            <tool>tool_name(param1, param2): description</tool>
        """
        tool_name = tool.name
        if compression_level == CompressionLevel.MAX:
            # Maximum compression: tool name only, no args, no description
            return f"<tool>{tool_name}</tool>"
        tool_arg_names = list(tool.parameters.get("properties", {}))
        tool_description = (tool.description or "").strip()
        if compression_level == CompressionLevel.HIGH:
            tool_description = ""
        elif tool_description and compression_level == CompressionLevel.MEDIUM:
            tool_description = tool_description.splitlines()[0].split(".")[0]
        tool_description = ": " + tool_description if tool_description else ""
        return f"<tool>{tool_name}({', '.join(tool_arg_names)}){tool_description}</tool>"

    def _is_built_in_tool(self, tool: Tool) -> bool:
        """Check if a tool is one of the built-in wrapper tools.

        Args:
            tool: The tool to check.

        Returns:
            True if the tool is a built-in wrapper tool, False otherwise.
        """
        return self._is_built_in_tool_name(tool.name)

    def _is_built_in_tool_name(self, tool_name: str) -> bool:
        """Check if a tool name refers to one of the built-in wrapper tools."""
        if self._cli_mode:
            return tool_name == self._help_tool_name
        return tool_name in self._built_in_tool_names

    def _is_hidden_tool_name(self, tool_name: str) -> bool:
        """Check if a built-in tool should be omitted from list_tools responses."""
        return tool_name in self._hidden_tool_names

    async def _get_backend_tools(self) -> dict[str, Tool]:
        """Retrieve backend tools from the proxy server, caching the result."""
        if self._all_tools is None:
            self._all_tools = await self._proxy_server.get_tools()
        return {name: tool for name, tool in self._all_tools.items() if not self._is_built_in_tool(tool)}

    async def _get_built_in_tools(self) -> dict[str, Tool]:
        """Retrieve built-in wrapper tools from the proxy server, caching the result."""
        if self._all_tools is None:
            self._all_tools = await self._proxy_server.get_tools()
        return {name: tool for name, tool in self._all_tools.items() if self._is_built_in_tool(tool)}

    async def _get_backend_tool(self, tool_name: str) -> Tool:
        """Retrieve a specific backend tool from the proxy server."""
        backend_tools = await self._get_backend_tools()
        tool = backend_tools.get(tool_name)
        if tool is None:
            available_tools = tuple(sorted(backend_tools))
            logger.error(f"Tool '{tool_name}' not found in backend tools. Available tools: {available_tools}")
            raise ToolNotFoundError(tool_name, available_tools)
        return tool

    async def _format_validation_error(self, tool_name: str, error_message: str) -> str:
        """Format a validation failure with the tool schema for client guidance."""
        tool_schema = await self.get_tool_schema(tool_name)
        return (
            f"Tool '{tool_name}' input validation failed: {error_message}\n\n"
            f"Here is the result of get_tool_schema('{tool_name}'):\n{tool_schema}"
        )

    def _is_validation_error_message(self, error_message: str) -> bool:
        """Return whether a tool error message appears to be an input validation failure."""
        lowered_message = error_message.lower()
        return "validation error" in lowered_message or "missing required argument" in lowered_message

    def _toonify_tool_result(self, tool_result: ToolResult) -> ToolResult:
        """Convert JSON text content blocks in a tool result to TOON format."""
        converted_content: list[ContentBlock] = []
        content_changed = False
        for content_block in tool_result.content:
            if isinstance(content_block, TextContent):
                converted_text = self._toonify_json_text(content_block.text)
                if converted_text != content_block.text:
                    content_changed = True
                    converted_content.append(TextContent(type="text", text=converted_text))
                    continue
            converted_content.append(content_block)
        if not content_changed:
            return tool_result
        return ToolResult(
            content=converted_content,
            structured_content=tool_result.structured_content,
            meta=tool_result.meta,
        )

    def _toonify_json_text(self, text: str) -> str:
        """Convert a JSON object/array string to TOON; pass through other text unchanged."""
        try:
            parsed = json.loads(text)
        except json.JSONDecodeError:
            return text
        if not isinstance(parsed, dict | list):
            return text
        return toons.dumps(parsed)

`init(proxy_server, compression_level, server_name=None, toonify=False, cli_mode=False, cli_name=None)` ¶

Initialize the CompressedTools middleware.

Parameters:

Name	Type	Description	Default
`proxy_server`	`FastMCP`	The MCP proxy server instance.	required
`compression_level`	`CompressionLevel`	The level of compression to apply to tool descriptions.	required
`server_name`	`str \| None`	Optional custom name prefix for tool names (will be sanitized and used as prefix).	`None`
`toonify`	`bool`	Whether to convert JSON text tool outputs to TOON format.	`False`
`cli_mode`	`bool`	Whether to run in CLI mode (exposes a single help tool instead of wrapper tools).	`False`
`cli_name`	`str \| None`	The CLI script name (used in CLI mode for help text).	`None`

Source code in mcp_compressor/tools.py

def __init__(
    self,
    proxy_server: FastMCP,
    compression_level: CompressionLevel,
    server_name: str | None = None,
    toonify: bool = False,
    cli_mode: bool = False,
    cli_name: str | None = None,
) -> None:
    """Initialize the CompressedTools middleware.

    Args:
        proxy_server: The MCP proxy server instance.
        compression_level: The level of compression to apply to tool descriptions.
        server_name: Optional custom name prefix for tool names (will be sanitized and used as prefix).
        toonify: Whether to convert JSON text tool outputs to TOON format.
        cli_mode: Whether to run in CLI mode (exposes a single help tool instead of wrapper tools).
        cli_name: The CLI script name (used in CLI mode for help text).
    """
    self._proxy_server = proxy_server
    self._compression_level = compression_level
    self._tool_name_prefix = f"{server_name}_" if server_name else ""
    self._server_description = f"the {server_name} toolset" if server_name else "this toolset"
    self._toonify = toonify
    self._cli_mode = cli_mode
    self._cli_name = cli_name or (server_name or "mcp")
    self._help_tool_name = sanitize_tool_name(f"{server_name}_help" if server_name else "help")
    self._get_schema_tool_name = sanitize_tool_name(f"{self._tool_name_prefix}get_tool_schema")
    self._invoke_tool_name = sanitize_tool_name(f"{self._tool_name_prefix}invoke_tool")
    self._invoke_tool_alias_name = sanitize_tool_name("invoke_tool")
    self._list_tools_name = sanitize_tool_name(f"{self._tool_name_prefix}list_tools")
    self._uncompressed_tools_resource_uri = "compressor://uncompressed-tools"
    self._hidden_tool_names: set[str] = set()
    self._built_in_tool_names = {self._get_schema_tool_name, self._invoke_tool_name}
    if self._invoke_tool_alias_name != self._invoke_tool_name:
        self._built_in_tool_names.add(self._invoke_tool_alias_name)
        self._hidden_tool_names.add(self._invoke_tool_alias_name)
    if self._compression_level == CompressionLevel.MAX:
        self._built_in_tool_names.add(self._list_tools_name)
    self._all_tools: dict[str, Tool] | None = None

`configure_server()` `async` ¶

Configure an MCP server with compressed tool wrappers.

In normal mode, registers get_tool_schema, invoke_tool, and optionally list_tools. In CLI mode, registers a single _help tool.

Source code in mcp_compressor/tools.py

async def configure_server(self) -> None:
    """Configure an MCP server with compressed tool wrappers.

    In normal mode, registers get_tool_schema, invoke_tool, and optionally list_tools.
    In CLI mode, registers a single <server_name>_help tool.
    """
    if self._cli_mode:

        async def help_tool() -> str:
            return await self._build_cli_description()

        help_tool.__doc__ = f"Get help for the '{self._cli_name}' CLI. Lists all available subcommands."
        self._proxy_server.tool(name=self._help_tool_name)(help_tool)
    else:
        # Create tool names with optional server name prefix
        self._proxy_server.tool(name=self._get_schema_tool_name)(self.get_tool_schema)
        self._proxy_server.tool(name=self._invoke_tool_name)(self.invoke_tool)
        if self._invoke_tool_alias_name != self._invoke_tool_name:
            self._proxy_server.tool(name=self._invoke_tool_alias_name)(self.invoke_tool)
        self._proxy_server.resource(
            uri=self._uncompressed_tools_resource_uri,
            description="The upstream server's original uncompressed tool list as JSON.",
            mime_type="application/json",
        )(self.list_uncompressed_tools)
        if self._compression_level == CompressionLevel.MAX:
            self._proxy_server.tool(name=self._list_tools_name)(self.list_tools)
    self._proxy_server.add_middleware(self)
    self._all_tools = None  # Reset cached tools, if any

`get_compression_stats()` `async` ¶

Get statistics about the compression of tool descriptions.

Computes the original backend schema size vs the compressed proxy tool size for all compression levels. Works identically in both normal and CLI mode — the only difference is which tools _get_built_in_tools() returns.

Returns:

Type	Description
`dict[str, Any]`	A dictionary containing statistics about the original and compressed tool description sizes.

Source code in mcp_compressor/tools.py

async def get_compression_stats(self) -> dict[str, Any]:
    """Get statistics about the compression of tool descriptions.

    Computes the original backend schema size vs the compressed proxy tool size
    for all compression levels. Works identically in both normal and CLI mode —
    the only difference is which tools _get_built_in_tools() returns.

    Returns:
        A dictionary containing statistics about the original and compressed tool description sizes.
    """
    backend_tools = await self._get_backend_tools()
    original_tool_count = len(backend_tools)
    original_schema_size = sum(
        len(json.dumps(tool.parameters)) + len(json.dumps(tool.output_schema)) + len(tool.description or "")
        for tool in backend_tools.values()
    )
    # Low/Medium/High/Max: always measured from the backend tools compressed at each level.
    # This is what a non-CLI agent would see in the get_tool_schema description,
    # and it gives meaningful differentiation between levels in both modes.
    compressed_tool_count = len(backend_tools)
    compressed_schema_sizes: dict[CompressionLevel | str, int] = {}

    for compression_level in [
        CompressionLevel.LOW,
        CompressionLevel.MEDIUM,
        CompressionLevel.HIGH,
        CompressionLevel.MAX,
    ]:
        compressed_schema_sizes[compression_level] = sum(
            len(self._format_tool_description(tool, compression_level)) for tool in backend_tools.values()
        )

    # "cli" key: the help tool description — what the agent sees in CLI mode.
    compressed_schema_sizes["cli"] = len(await self._build_cli_description())
    return {
        "original_tool_count": original_tool_count,
        "compressed_tool_count": compressed_tool_count,
        "original_schema_size": original_schema_size,
        "compressed_schema_sizes": compressed_schema_sizes,
    }

`get_tool_schema(tool_name)` `async` ¶

Get the input schema for a specific tool from {server_description}.

Available tools are: {tool_descriptions}

Parameters:

Name	Type	Description	Default
`tool_name`	`str`	The name of the tool to get the schema for.	required

Returns:

Type	Description
`str`	The input schema for the specified tool.

Raises:

Type	Description
`ValueError`	If the tool name is not found in the server.

Source code in mcp_compressor/tools.py

async def get_tool_schema(self, tool_name: str) -> str:
    """Get the input schema for a specific tool from {server_description}.

    Available tools are:
    {tool_descriptions}

    Args:
        tool_name: The name of the tool to get the schema for.

    Returns:
        The input schema for the specified tool.

    Raises:
        ValueError: If the tool name is not found in the server.
    """
    tool = await self._get_backend_tool(tool_name)
    tool_description = self._format_tool_description(tool, CompressionLevel.LOW)
    return tool_description + "\n\n" + json.dumps(tool.parameters, indent=2)

`invoke_tool(tool_name, tool_input=None, quiet=False)` `async` ¶

Invoke a tool from {server_description}.

Parameters:

Name	Type	Description	Default
`tool_name`	`str`	The name of the tool to invoke.	required
`tool_input`	`dict[str, Any] \| None`	The input to the tool. Schemas can be retrieved using the appropriate `get_tool_schema` function.	`None`
`quiet`	`bool`	If true, truncates large tool outputs for successful invocations. This is useful for reducing token consumption when the output is not needed. Full responses will always be returned for tool errors.	`False`

Returns:

Type	Description
`Any`	The output from the tool.

Source code in mcp_compressor/tools.py

async def invoke_tool(self, tool_name: str, tool_input: dict[str, Any] | None = None, quiet: bool = False) -> Any:
    """Invoke a tool from {server_description}.

    Args:
        tool_name: The name of the tool to invoke.
        tool_input: The input to the tool. Schemas can be retrieved using the appropriate `get_tool_schema`
            function.
        quiet: If true, truncates large tool outputs for successful invocations. This is useful for reducing token
            consumption when the output is not needed. Full responses will always be returned for tool errors.

    Returns:
        The output from the tool.
    """
    tool = await self._get_backend_tool(tool_name)
    try:
        tool_result = await tool.run(tool_input or {})
    except ValidationError as exc:
        raise ToolError(await self._format_validation_error(tool_name, str(exc))) from exc
    except ToolError as exc:
        if self._is_validation_error_message(str(exc)):
            raise ToolError(await self._format_validation_error(tool_name, str(exc))) from exc
        raise
    if self._toonify:
        tool_result = self._toonify_tool_result(tool_result)
    if quiet:
        if len(tool_result.content) == 1 and isinstance(tool_result.content[0], TextContent):
            return_text = tool_result.content[0].text
            if len(return_text) < QUIET_MODE_THRESHOLD:
                return tool_result
            preview_length = QUIET_MODE_THRESHOLD // 2
            return_text = (
                return_text[:preview_length]
                + "\n...\n(truncated due to quiet mode)\n...\n"
                + return_text[-preview_length:]
            )
        else:
            return_text = f"Successfully executed tool '{tool.name}' without output."
        return ToolResult(content=[TextContent(type="text", text=return_text)])
    return tool_result

`list_tools()` `async` ¶

List all available tools in {server_description}.

Returns:

Type	Description
`str`	A formatted string listing tool names and brief descriptions.

Source code in mcp_compressor/tools.py

async def list_tools(self) -> str:
    """List all available tools in {server_description}.

    Returns:
        A formatted string listing tool names and brief descriptions.
    """
    return await self._get_tool_descriptions(CompressionLevel.MEDIUM)

`list_uncompressed_tools()` `async` ¶

Return the upstream server's original list_tools payload as JSON.

This hidden helper is intended for MCP-aware clients that need the backend server's uncompressed tool inventory, including the same descriptions and schemas exposed by the upstream list_tools endpoint.

Returns:

Type	Description
`str`	A JSON array matching the upstream server's list_tools response.

Source code in mcp_compressor/tools.py

async def list_uncompressed_tools(self) -> str:
    """Return the upstream server's original list_tools payload as JSON.

    This hidden helper is intended for MCP-aware clients that need the backend server's uncompressed tool
    inventory, including the same descriptions and schemas exposed by the upstream list_tools endpoint.

    Returns:
        A JSON array matching the upstream server's list_tools response.
    """
    tools = []
    for tool in (await self._get_backend_tools()).values():
        tool_data = tool.model_dump(mode="json")
        tools.append({
            "name": tool_data["name"],
            "title": tool_data["title"],
            "description": tool_data["description"],
            "inputSchema": tool_data["parameters"],
            "outputSchema": tool_data["output_schema"],
            "icons": tool_data["icons"],
            "annotations": tool_data["annotations"],
            "meta": tool_data["meta"],
            "execution": tool_data["execution"],
        })
    return json.dumps(tools, indent=2)

`on_call_tool(context, call_next)` `async` ¶

Middleware to clean up tool invocation arguments to invoke_tool and route to the underlying tool.

Parameters:

Name	Type	Description	Default
`context`	`MiddlewareContext[CallToolRequestParams]`	The middleware context containing the call request.	required
`call_next`	`CallNext[CallToolRequestParams, ToolResult]`	The next middleware or handler in the chain.	required

Returns:

Type	Description
`ToolResult`	The result from calling the underlying tool.

Source code in mcp_compressor/tools.py

async def on_call_tool(
    self,
    context: MiddlewareContext[CallToolRequestParams],
    call_next: CallNext[CallToolRequestParams, ToolResult],
) -> ToolResult:
    """Middleware to clean up tool invocation arguments to invoke_tool and route to the underlying tool.

    Args:
        context: The middleware context containing the call request.
        call_next: The next middleware or handler in the chain.

    Returns:
        The result from calling the underlying tool.
    """
    tool_args = context.message.arguments
    if not context.message.name.endswith("invoke_tool") or not tool_args:
        result = await call_next(context)
        if self._toonify and not self._is_built_in_tool_name(context.message.name):
            return self._toonify_tool_result(result)
        return result

    if "tool_input" not in tool_args or tool_args["tool_input"] is None:
        tool_input = {k: v for k, v in tool_args.items() if k not in ["tool_name", "quiet"]}
    else:
        tool_input = tool_args["tool_input"]
    return await self.invoke_tool(
        tool_name=tool_args["tool_name"],
        tool_input=tool_input,
        quiet=tool_args.get("quiet", False),
    )

`on_list_tools(context, call_next)` `async` ¶

Middleware to inject compressed tool descriptions and suppress backend tools.

In normal mode, updates get_tool_schema's description with the tool list. In CLI mode, updates the help tool's description with the full CLI help text.

Returns:

Type	Description
`Sequence[Tool]`	The sequence of built-in wrapper tools with updated descriptions.

Source code in mcp_compressor/tools.py

async def on_list_tools(
    self, context: MiddlewareContext[ListToolsRequest], call_next: CallNext[ListToolsRequest, Sequence[Tool]]
) -> Sequence[Tool]:
    """Middleware to inject compressed tool descriptions and suppress backend tools.

    In normal mode, updates get_tool_schema's description with the tool list.
    In CLI mode, updates the help tool's description with the full CLI help text.

    Returns:
        The sequence of built-in wrapper tools with updated descriptions.
    """
    built_in_tools = [
        tool for tool in (await self._get_built_in_tools()).values() if not self._is_hidden_tool_name(tool.name)
    ]
    if self._cli_mode:
        description = await self._build_cli_description()
        for tool in built_in_tools:
            tool.description = description
        return built_in_tools
    prepared_tools = []
    for tool in built_in_tools:
        logger.info(f"Preparing tool: {tool.name}")
        prepared_tools.append(tool)
        tool.description = cast(str, tool.description).format(
            tool_descriptions=await self._get_tool_descriptions(self._compression_level),
            server_description=self._server_description,
        )
    return prepared_tools

`ToolNotFoundError` ¶

Bases: ValueError

Exception raised when a requested tool is not found in the backend MCP server.

Source code in mcp_compressor/tools.py

class ToolNotFoundError(ValueError):
    """Exception raised when a requested tool is not found in the backend MCP server."""

    def __init__(self, tool_name: str, available_tools: Sequence[str]) -> None:
        self.tool_name = tool_name
        self.available_tools = tuple(available_tools)
        available_tools_text = ", ".join(self.available_tools) if self.available_tools else "(none)"
        super().__init__(f"Tool '{tool_name}' not found in backend MCP server. Available tools: {available_tools_text}")

`sanitize_tool_name(name)` ¶

Sanitize a tool name to conform to MCP tool name specifications.

Tool names must: - Be between 1 and 128 characters (inclusive) - Only contain: A-Z, a-z, 0-9, underscore (_), hyphen (-), and dot (.) - Not contain spaces, commas, or other special characters

Parameters:

Name	Type	Description	Default
`name`	`str`	The raw tool name to sanitize.	required

Returns:

Type	Description
`str`	A sanitized tool name conforming to MCP specifications.

Source code in mcp_compressor/tools.py

def sanitize_tool_name(name: str) -> str:
    """Sanitize a tool name to conform to MCP tool name specifications.

    Tool names must:
    - Be between 1 and 128 characters (inclusive)
    - Only contain: A-Z, a-z, 0-9, underscore (_), hyphen (-), and dot (.)
    - Not contain spaces, commas, or other special characters

    Args:
        name: The raw tool name to sanitize.

    Returns:
        A sanitized tool name conforming to MCP specifications.
    """
    # Replace spaces and invalid characters with underscores
    sanitized = re.sub(r"[^A-Za-z0-9_\-.]", "_", name).lower()

    # Ensure the name is not empty after sanitization
    if not sanitized:
        raise ValueError("Tool name must contain at least one valid character after sanitization.")  # noqa: TRY003

    # Truncate to 128 characters if needed
    return sanitized[:128]

Types Module¶

Type definitions for MCP Compressor.

This module defines enumerations and type aliases used throughout the MCP Compressor package.

`ToolResultBlock = str | Audio | File | Image` `module-attribute` ¶

Type alias for possible tool result content blocks (text, audio, file, or image).

`TransportType = SSETransport | StdioTransport | StreamableHttpTransport` `module-attribute` ¶

Type alias for supported MCP transport types (SSE, stdio, or streamable HTTP).

`CompressionLevel` ¶

Bases: str, Enum

Compression levels for tool descriptions in the wrapped MCP server.

Higher compression levels provide less verbose tool descriptions, reducing token usage. Lower compression levels provide more detailed information upfront.

Attributes:

Name	Type	Description
`MAX`		Maximum compression - exposes a list_tools function for viewing all tools.
`HIGH`		High compression - only tool names and parameter names, no descriptions.
`MEDIUM`		Medium compression - first sentence of tool descriptions only.
`LOW`		Low compression - complete tool descriptions and schemas.

Source code in mcp_compressor/types.py

class CompressionLevel(str, Enum):
    """Compression levels for tool descriptions in the wrapped MCP server.

    Higher compression levels provide less verbose tool descriptions, reducing token usage.
    Lower compression levels provide more detailed information upfront.

    Attributes:
        MAX: Maximum compression - exposes a list_tools function for viewing all tools.
        HIGH: High compression - only tool names and parameter names, no descriptions.
        MEDIUM: Medium compression - first sentence of tool descriptions only.
        LOW: Low compression - complete tool descriptions and schemas.
    """

    MAX = "max"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"

`LogLevel` ¶

Bases: str, Enum

Logging levels for the MCP Compressor server and wrapped MCP servers.

Source code in mcp_compressor/types.py

class LogLevel(str, Enum):
    """Logging levels for the MCP Compressor server and wrapped MCP servers."""

    DEBUG = "DEBUG"
    INFO = "INFO"
    WARNING = "WARNING"
    ERROR = "ERROR"
    CRITICAL = "CRITICAL"

API Reference¶

Main Module¶

clear_oauth(all_tokens=False) ¶

entrypoint() ¶

main(command_or_url_list, cwd=None, env_list=None, header_list=None, timeout=10.0, compression_level=CompressionLevel.MEDIUM, server_name=None, log_level=LogLevel.WARNING, toonify=False, cli_mode=False, cli_port=None) ¶

Tools Module¶

CompressedTools ¶

__init__(proxy_server, compression_level, server_name=None, toonify=False, cli_mode=False, cli_name=None) ¶

configure_server() async ¶

get_compression_stats() async ¶

get_tool_schema(tool_name) async ¶

invoke_tool(tool_name, tool_input=None, quiet=False) async ¶

list_tools() async ¶

list_uncompressed_tools() async ¶

on_call_tool(context, call_next) async ¶

on_list_tools(context, call_next) async ¶

ToolNotFoundError ¶

sanitize_tool_name(name) ¶

Types Module¶

ToolResultBlock = str | Audio | File | Image module-attribute ¶

TransportType = SSETransport | StdioTransport | StreamableHttpTransport module-attribute ¶

CompressionLevel ¶

LogLevel ¶

`clear_oauth(all_tokens=False)` ¶

`entrypoint()` ¶

`main(command_or_url_list, cwd=None, env_list=None, header_list=None, timeout=10.0, compression_level=CompressionLevel.MEDIUM, server_name=None, log_level=LogLevel.WARNING, toonify=False, cli_mode=False, cli_port=None)` ¶

`CompressedTools` ¶

`init(proxy_server, compression_level, server_name=None, toonify=False, cli_mode=False, cli_name=None)` ¶

`configure_server()` `async` ¶

`get_compression_stats()` `async` ¶

`get_tool_schema(tool_name)` `async` ¶

`invoke_tool(tool_name, tool_input=None, quiet=False)` `async` ¶

`list_tools()` `async` ¶

`list_uncompressed_tools()` `async` ¶

`on_call_tool(context, call_next)` `async` ¶

`on_list_tools(context, call_next)` `async` ¶

`ToolNotFoundError` ¶

`sanitize_tool_name(name)` ¶

`ToolResultBlock = str | Audio | File | Image` `module-attribute` ¶

`TransportType = SSETransport | StdioTransport | StreamableHttpTransport` `module-attribute` ¶

`CompressionLevel` ¶

`LogLevel` ¶