Skip to content

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.

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)

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 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@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,
):
    """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",))

    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,
        )
    )

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. Instead of exposing tools directly with full schemas, it provides two or three 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)

The compression level determines how much information is included in the get_tool_schema tool description.

Source code in mcp_compressor/tools.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
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.
    Instead of exposing tools directly with full schemas, it provides two or three 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)

    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
    ) -> 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).
        """
        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._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.
        """
        pass

    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:
            return await call_next(context)

        if "tool_input" not in tool_args or not tool_args["tool_input"]:
            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"]
        tool = await self._get_backend_tool(tool_args["tool_name"])
        tool_result = await tool.run(tool_input)
        if tool_args.get("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 on_list_tools(
        self, context: MiddlewareContext[ListToolsRequest], call_next: CallNext[ListToolsRequest, Sequence[Tool]]
    ) -> Sequence[Tool]:
        """Middleware to inject compressed tool descriptions into the get_tool_schema tool and suppress backend tools.

        This updates the get_tool_schema tool's description to include the list of available tools at the appropriate
        compression level.

        Returns:
            The sequence of tools with updated descriptions.
        """
        prepared_tools = []
        for tool_name, tool in (await self._get_built_in_tools()).items():
            logger.info(f"Preparing tool: {tool_name}")
            prepared_tools.append(tool)
            tool.description = cast(str, tool.description)
            tool.description = 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.

        This initializes the tools from the underlying server and registers the wrapper tools (get_tool_schema,
        invoke_tool, and optionally list_tools) on the provided server.

        Returns:
            The number of tools registered on the server.
        """
        # Create tool names with optional server name prefix
        get_schema_name = sanitize_tool_name(f"{self._tool_name_prefix}get_tool_schema")
        invoke_tool_name = sanitize_tool_name(f"{self._tool_name_prefix}invoke_tool")
        list_tools_name = sanitize_tool_name(f"{self._tool_name_prefix}list_tools")

        self._proxy_server.tool(name=get_schema_name)(self.get_tool_schema)
        self._proxy_server.tool(name=invoke_tool_name)(self.invoke_tool)
        if self._compression_level == CompressionLevel.MAX:
            self._proxy_server.tool(name=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.

        Returns:
            A dictionary containing statistics about the original and compressed tool description sizes.
        """
        backend_tools = await self._get_backend_tools()
        proxy_tools = await self._get_built_in_tools()
        original_tool_count = len(backend_tools)
        compressed_tool_count = len(proxy_tools)
        original_schema_size = 0
        for tool in backend_tools.values():
            original_schema_size += len(json.dumps(tool.parameters))
            original_schema_size += len(json.dumps(tool.output_schema))
            original_schema_size += len(tool.description or "")
        compressed_schema_sizes = {}
        for compression_level in [
            CompressionLevel.LOW,
            CompressionLevel.MEDIUM,
            CompressionLevel.HIGH,
            CompressionLevel.MAX,
        ]:
            compressed_size = 0
            for tool in proxy_tools.values():
                compressed_size += len(json.dumps(tool.parameters))
                compressed_size += len(json.dumps(tool.output_schema))
                compressed_size += len(
                    cast(str, tool.description).format(
                        tool_descriptions=await self._get_tool_descriptions(compression_level),
                        server_description=self._server_description,
                    )
                )
            compressed_schema_sizes[compression_level] = compressed_size
        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 _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
        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 any(tool.name.endswith(suffix) for suffix in ("get_tool_schema", "invoke_tool", "list_tools"))

    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:
            logger.error(f"Tool '{tool_name}' not found in backend tools.")
            raise ToolNotFoundError(tool_name)
        return tool

__init__(proxy_server, compression_level, server_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
Source code in mcp_compressor/tools.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(
    self, proxy_server: FastMCP, compression_level: CompressionLevel, server_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).
    """
    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._all_tools: dict[str, Tool] | None = None

configure_server() async

Configure an MCP server with compressed tool wrappers.

This initializes the tools from the underlying server and registers the wrapper tools (get_tool_schema, invoke_tool, and optionally list_tools) on the provided server.

Returns:

Type Description
None

The number of tools registered on the server.
Source code in mcp_compressor/tools.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
async def configure_server(self) -> None:
    """Configure an MCP server with compressed tool wrappers.

    This initializes the tools from the underlying server and registers the wrapper tools (get_tool_schema,
    invoke_tool, and optionally list_tools) on the provided server.

    Returns:
        The number of tools registered on the server.
    """
    # Create tool names with optional server name prefix
    get_schema_name = sanitize_tool_name(f"{self._tool_name_prefix}get_tool_schema")
    invoke_tool_name = sanitize_tool_name(f"{self._tool_name_prefix}invoke_tool")
    list_tools_name = sanitize_tool_name(f"{self._tool_name_prefix}list_tools")

    self._proxy_server.tool(name=get_schema_name)(self.get_tool_schema)
    self._proxy_server.tool(name=invoke_tool_name)(self.invoke_tool)
    if self._compression_level == CompressionLevel.MAX:
        self._proxy_server.tool(name=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.

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 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
async def get_compression_stats(self) -> dict[str, Any]:
    """Get statistics about the compression of tool descriptions.

    Returns:
        A dictionary containing statistics about the original and compressed tool description sizes.
    """
    backend_tools = await self._get_backend_tools()
    proxy_tools = await self._get_built_in_tools()
    original_tool_count = len(backend_tools)
    compressed_tool_count = len(proxy_tools)
    original_schema_size = 0
    for tool in backend_tools.values():
        original_schema_size += len(json.dumps(tool.parameters))
        original_schema_size += len(json.dumps(tool.output_schema))
        original_schema_size += len(tool.description or "")
    compressed_schema_sizes = {}
    for compression_level in [
        CompressionLevel.LOW,
        CompressionLevel.MEDIUM,
        CompressionLevel.HIGH,
        CompressionLevel.MAX,
    ]:
        compressed_size = 0
        for tool in proxy_tools.values():
            compressed_size += len(json.dumps(tool.parameters))
            compressed_size += len(json.dumps(tool.output_schema))
            compressed_size += len(
                cast(str, tool.description).format(
                    tool_descriptions=await self._get_tool_descriptions(compression_level),
                    server_description=self._server_description,
                )
            )
        compressed_schema_sizes[compression_level] = compressed_size
    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 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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.
    """
    pass

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 
61
62
63
64
65
66
67
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)

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 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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:
        return await call_next(context)

    if "tool_input" not in tool_args or not tool_args["tool_input"]:
        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"]
    tool = await self._get_backend_tool(tool_args["tool_name"])
    tool_result = await tool.run(tool_input)
    if tool_args.get("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

on_list_tools(context, call_next) async

Middleware to inject compressed tool descriptions into the get_tool_schema tool and suppress backend tools.

This updates the get_tool_schema tool's description to include the list of available tools at the appropriate compression level.

Returns:

Type Description
Sequence[Tool]

The sequence of tools with updated descriptions.
Source code in mcp_compressor/tools.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
async def on_list_tools(
    self, context: MiddlewareContext[ListToolsRequest], call_next: CallNext[ListToolsRequest, Sequence[Tool]]
) -> Sequence[Tool]:
    """Middleware to inject compressed tool descriptions into the get_tool_schema tool and suppress backend tools.

    This updates the get_tool_schema tool's description to include the list of available tools at the appropriate
    compression level.

    Returns:
        The sequence of tools with updated descriptions.
    """
    prepared_tools = []
    for tool_name, tool in (await self._get_built_in_tools()).items():
        logger.info(f"Preparing tool: {tool_name}")
        prepared_tools.append(tool)
        tool.description = cast(str, tool.description)
        tool.description = 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 
25
26
27
28
29
30
class ToolNotFoundError(ValueError):
    """Exception raised when a requested tool is not found in the backend MCP server."""

    def __init__(self, tool_name: str) -> None:
        super().__init__(f"Tool '{tool_name}' not found in backend MCP server.")
        self.tool_name = tool_name

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 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
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 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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 
13
14
15
16
17
18
19
20
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"