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 |
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 ¶
Returns:
Type | Description |
---|---|
Server | The underlying low-level server instance. You can use this to register additional handlers. |
session_manager ¶
Returns:
Type | Description |
---|---|
StreamableHTTPSessionManager | The underlying session manager instance. You only need to use this if you want to override the |
commands ¶
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
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. |
list_tools_handler ¶
call_tool_handler ¶
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 |
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 |