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(_KEYRING_SERVICE, _KEYRING_USERNAME)
                print(f"  keyring entry: {_KEYRING_SERVICE} / {_KEYRING_USERNAME}")
        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(ctx, 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.ERROR, toonify=False, cli_mode=False, just_bash=False, cli_port=None, include_tools=None, exclude_tools=None, version=False)` ¶

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(
    ctx: typer.Context,
    command_or_url_list: Annotated[
        list[str],
        typer.Argument(
            ...,
            metavar="COMMAND_OR_URL",
            help=(
                "The backend to wrap: either a remote MCP URL, a stdio command plus arguments, or an MCP config "
                "JSON string with one or more servers. Example stdio usage: 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.ERROR,
    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,
    just_bash: Annotated[
        bool,
        typer.Option(
            ...,
            "--just-bash",
            help=(
                "Start in just-bash mode: expose a single 'bash' MCP tool powered by "
                "just-bash, with all backend server tools available as custom commands. "
                "--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,
    include_tools: Annotated[
        str | None,
        typer.Option(
            ...,
            "--include-tools",
            help=("Comma-separated list of wrapped server tool names to expose. If omitted, all tools are included."),
        ),
    ] = None,
    exclude_tools: Annotated[
        str | None,
        typer.Option(
            ...,
            "--exclude-tools",
            help="Comma-separated list of wrapped server tool names to hide.",
        ),
    ] = None,
    version: Annotated[
        bool,
        typer.Option(
            "--version",
            "-V",
            help="Show the version and exit.",
            callback=_version_callback,
            is_eager=True,
            expose_value=False,
        ),
    ] = False,
):
    """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.
    """
    configure_logging(log_level)

    resolved_server_name = server_name
    try:
        parsed_config = _parse_mcp_config_json(command_or_url_list)
    except ValueError as exc:
        raise typer.BadParameter(str(exc), param_hint="'COMMAND_OR_URL'") from exc

    if parsed_config is not None:
        _check_conflicting_config_options(ctx)
        resolved_server_name = _resolve_config_server_name(parsed_config, server_name, cli_mode)

    if (
        (cli_mode or just_bash)
        and resolved_server_name is None
        and (parsed_config is None or len(parsed_config.mcpServers) == 1)
    ):
        raise typer.BadParameter(
            "--server-name is required when using --cli-mode or --just-bash.", param_hint="'--server-name'"
        )
    if compression_level == CompressionLevel.MAX and resolved_server_name is None:
        raise typer.BadParameter(
            "--server-name is required when using --compression-level=max.", param_hint="'--server-name'"
        )

    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=resolved_server_name,
            log_level=log_level,
            toonify=toonify or cli_mode or just_bash,
            cli_mode=cli_mode,
            just_bash=just_bash,
            cli_port=cli_port,
            include_tools=_parse_tool_name_list(include_tools),
            exclude_tools=_parse_tool_name_list(exclude_tools),
        )
    )

Tools Module¶

Tool compression helpers built on FastMCP v3 transforms.

This module provides a transform-first implementation that replaces the visible tool catalog with a compressed wrapper interface while keeping backend tools available for passthrough access.

`CompressedTools` ¶

Bases: CatalogTransform

Transform that replaces the tool catalog with compressed wrapper tools.

In normal mode it exposes 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 exposes a resource (compressor://uncompressed-tools) that returns the upstream server's original list_tools payload in machine-readable JSON form.

In CLI mode it exposes a single help tool (_help) instead of the wrapper tool catalog.