Skip to content

API reference


CommandMCPServer

An MCP server that can be used to run Python CLIs, backed by Starlette and Uvicorn. Example usage:

from pycli_mcp import CommandMCPServer

from mypkg.cli import cmd

server = CommandMCPServer(commands=[cmd], stateless=True)
server.run()

Parameters:

Name Type Description Default
commands Sequence[Any]

The commands to expose as MCP tools.

required

Other Parameters:

Name Type Description
event_store EventStore | None

Optional event store that allows clients to reconnect and receive missed events. If None, sessions are still tracked but not resumable.

stateless bool

Whether to create a completely fresh transport for each request with no session tracking or state persistence between requests.

**app_settings Any

Additional settings to pass to the Starlette application.

Source code in src/pycli_mcp/server.py
class CommandMCPServer:
    """
    An MCP server that can be used to run Python CLIs, backed by [Starlette](https://github.com/encode/starlette)
    and [Uvicorn](https://github.com/encode/uvicorn). Example usage:

    ```python
    from pycli_mcp import CommandMCPServer

    from mypkg.cli import cmd

    server = CommandMCPServer(commands=[cmd], stateless=True)
    server.run()
    ```

    Parameters:
        commands: The commands to expose as MCP tools.

    Other parameters:
        event_store: Optional [event store](https://github.com/modelcontextprotocol/python-sdk/blob/v1.9.4/src/mcp/server/streamable_http.py#L79)
            that allows clients to reconnect and receive missed events. If `None`, sessions are still tracked but not
            resumable.
        stateless: Whether to create a completely fresh transport for each request with no session tracking or state
            persistence between requests.
        **app_settings: Additional settings to pass to the Starlette [application][starlette.applications.Starlette].
    """

    def __init__(
        self,
        commands: Sequence[Any],
        *,
        event_store: EventStore | None = None,
        stateless: bool = False,
        **app_settings: Any,
    ) -> None:
        self.__command_queries = [c if isinstance(c, CommandQuery) else CommandQuery(c) for c in commands]
        self.__app_settings = app_settings
        self.__server: Server = Server("pycli_mcp")
        self.__session_manager = StreamableHTTPSessionManager(
            app=self.__server,
            event_store=event_store,
            stateless=stateless,
            json_response=True,
        )

        # Register handlers
        self.__server.request_handlers[ListToolsRequest] = self.list_tools_handler
        self.__server.request_handlers[CallToolRequest] = self.call_tool_handler

    @property
    def server(self) -> Server:
        """
        Returns:
            The underlying [low-level server](https://github.com/modelcontextprotocol/python-sdk/blob/v1.9.4/src/mcp/server/lowlevel/server.py)
                instance. You can use this to register additional handlers.
        """
        return self.__server

    @property
    def session_manager(self) -> StreamableHTTPSessionManager:
        """
        Returns:
            The underlying [session manager](https://github.com/modelcontextprotocol/python-sdk/blob/v1.9.4/src/mcp/server/streamable_http_manager.py#L29)
                instance. You only need to use this if you want to override the `lifetime` context manager
        """
        return self.__session_manager

    @cached_property
    def commands(self) -> dict[str, Command]:
        """
        Returns:
            Dictionary used internally to store metadata about the exposed commands. Although it should not be modified,
                the keys are the available MCP tool names and useful to know when overriding the default handlers.
        """
        commands: dict[str, Command] = {}
        for query in self.__command_queries:
            for metadata in query:
                tool_name = metadata.path.replace(" ", ".").replace("-", "_")
                tool = Tool(
                    name=tool_name,
                    description=metadata.schema["description"],
                    inputSchema=metadata.schema,
                )
                commands[tool_name] = Command(metadata, tool)

        return commands

    @cached_property
    def routes(self) -> list[Mount]:
        """
        This would only be used directly if you want to add more routes in addition to the default `/mcp` route.

        Returns:
            The [routes](https://www.starlette.io/routing/#http-routing) to mount in the Starlette
                [application][starlette.applications.Starlette].
        """
        return [Mount("/mcp", app=self.session_manager.handle_request)]

    @asynccontextmanager
    async def lifespan(self, app: Starlette) -> AsyncIterator[None]:  # noqa: ARG002
        """
        The default lifespan context manager used by the Starlette [application][starlette.applications.Starlette].
        """
        async with self.session_manager.run():
            yield

    def list_command_tools(self) -> list[Tool]:
        """
        This would only be used directly if you want to override the handler for the `ListToolsRequest`.

        Returns:
            The MCP tools for the commands.
        """
        return [command.tool for command in self.commands.values()]

    async def list_tools_handler(self, _: ListToolsRequest) -> ServerResult:
        """
        The default handler for the `ListToolsRequest`.
        """
        return ServerResult(ListToolsResult(tools=self.list_command_tools()))

    async def call_tool_handler(self, req: CallToolRequest) -> ServerResult:
        """
        The default handler for the `CallToolRequest`.
        """
        command = self.commands[req.params.name].metadata.construct(req.params.arguments)
        env_vars = dict(os.environ)
        env_vars["PYCLI_MCP_TOOL_NAME"] = req.params.name

        try:
            process = subprocess.run(  # noqa: PLW1510
                command,
                encoding="utf-8",
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                env=env_vars,
            )
        # This can happen if the command is not found
        except subprocess.CalledProcessError as e:
            return ServerResult(CallToolResult(content=[TextContent(type="text", text=str(e))], isError=True))

        if process.returncode:
            msg = f"{process.stdout}\nThis command exited with non-zero exit code `{process.returncode}`: {command}"
            return ServerResult(CallToolResult(content=[TextContent(type="text", text=msg)], isError=True))

        return ServerResult(CallToolResult(content=[TextContent(type="text", text=process.stdout)]))

    def run(self, **kwargs: Any) -> None:
        """
        Other parameters:
            **kwargs: Additional settings to pass to the [`uvicorn.run`](https://www.uvicorn.org/#uvicornrun) function.
        """
        app_settings = self.__app_settings.copy()
        app_settings["routes"] = self.routes
        app_settings.setdefault("lifespan", self.lifespan)
        app = Starlette(**app_settings)
        uvicorn.run(app, **kwargs)

server

server: Server

Returns:

Type Description
Server

The underlying low-level server instance. You can use this to register additional handlers.

session_manager

session_manager: StreamableHTTPSessionManager

Returns:

Type Description
StreamableHTTPSessionManager

The underlying session manager instance. You only need to use this if you want to override the lifetime context manager

commands

commands: dict[str, Command]

Returns:

Type Description
dict[str, Command]

Dictionary used internally to store metadata about the exposed commands. Although it should not be modified, the keys are the available MCP tool names and useful to know when overriding the default handlers.

routes

routes: list[Mount]

This would only be used directly if you want to add more routes in addition to the default /mcp route.

Returns:

Type Description
list[Mount]

The routes to mount in the Starlette application.

lifespan

lifespan(app: Starlette) -> AsyncIterator[None]

The default lifespan context manager used by the Starlette application.

Source code in src/pycli_mcp/server.py
@asynccontextmanager
async def lifespan(self, app: Starlette) -> AsyncIterator[None]:  # noqa: ARG002
    """
    The default lifespan context manager used by the Starlette [application][starlette.applications.Starlette].
    """
    async with self.session_manager.run():
        yield

list_command_tools

list_command_tools() -> list[Tool]

This would only be used directly if you want to override the handler for the ListToolsRequest.

Returns:

Type Description
list[Tool]

The MCP tools for the commands.

Source code in src/pycli_mcp/server.py
def list_command_tools(self) -> list[Tool]:
    """
    This would only be used directly if you want to override the handler for the `ListToolsRequest`.

    Returns:
        The MCP tools for the commands.
    """
    return [command.tool for command in self.commands.values()]

list_tools_handler

list_tools_handler(_: ListToolsRequest) -> ServerResult

The default handler for the ListToolsRequest.

Source code in src/pycli_mcp/server.py
async def list_tools_handler(self, _: ListToolsRequest) -> ServerResult:
    """
    The default handler for the `ListToolsRequest`.
    """
    return ServerResult(ListToolsResult(tools=self.list_command_tools()))

call_tool_handler

call_tool_handler(req: CallToolRequest) -> ServerResult

The default handler for the CallToolRequest.

Source code in src/pycli_mcp/server.py
async def call_tool_handler(self, req: CallToolRequest) -> ServerResult:
    """
    The default handler for the `CallToolRequest`.
    """
    command = self.commands[req.params.name].metadata.construct(req.params.arguments)
    env_vars = dict(os.environ)
    env_vars["PYCLI_MCP_TOOL_NAME"] = req.params.name

    try:
        process = subprocess.run(  # noqa: PLW1510
            command,
            encoding="utf-8",
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            env=env_vars,
        )
    # This can happen if the command is not found
    except subprocess.CalledProcessError as e:
        return ServerResult(CallToolResult(content=[TextContent(type="text", text=str(e))], isError=True))

    if process.returncode:
        msg = f"{process.stdout}\nThis command exited with non-zero exit code `{process.returncode}`: {command}"
        return ServerResult(CallToolResult(content=[TextContent(type="text", text=msg)], isError=True))

    return ServerResult(CallToolResult(content=[TextContent(type="text", text=process.stdout)]))

run

run(**kwargs: Any) -> None

Other Parameters:

Name Type Description
**kwargs Any

Additional settings to pass to the uvicorn.run function.

Source code in src/pycli_mcp/server.py
def run(self, **kwargs: Any) -> None:
    """
    Other parameters:
        **kwargs: Additional settings to pass to the [`uvicorn.run`](https://www.uvicorn.org/#uvicornrun) function.
    """
    app_settings = self.__app_settings.copy()
    app_settings["routes"] = self.routes
    app_settings.setdefault("lifespan", self.lifespan)
    app = Starlette(**app_settings)
    uvicorn.run(app, **kwargs)

CommandQuery

A wrapper around a root command object that influences the collection behavior. Example usage:

from pycli_mcp import CommandMCPServer, CommandQuery

from mypkg.cli import cmd

# Only expose the `foo` subcommand
query = CommandQuery(cmd, include=r"^foo$")
server = CommandMCPServer(commands=[query])
server.run()

Parameters:

Name Type Description Default
command Any

The command to inspect.

required
aggregate Literal['root', 'group', 'none'] | None

The level of aggregation to use.

None
name str | None

The expected name of the root command.

None
include str | Pattern | None

A regular expression to include in the query.

None
exclude str | Pattern | None

A regular expression to exclude in the query.

None
strict_types bool

Whether to error on unknown types.

False