Plugin API

This module contains the main plugin class for the Litestar MCP Plugin.

LitestarMCP

class litestar_mcp.plugin.LitestarMCP

Bases: InitPluginProtocol, CLIPlugin

Litestar plugin for Model Context Protocol integration.

The main plugin class that implements litestar.plugins.InitPluginProtocol. It discovers routes marked with mcp_tool, mcp_resource, or mcp_prompt in their opt dictionary and exposes them through the MCP Streamable HTTP transport surface. Standalone prompt callables decorated with mcp_prompt() can be passed via the prompts constructor argument.

__init__(config=None, prompts=None)

Initialize the MCP plugin.

Parameters:

• config (Optional[MCPConfig]) -- Plugin configuration. Defaults to MCPConfig().

• prompts (Optional[Sequence[Callable[..., Any]]]) -- Optional sequence of standalone prompt functions decorated with @mcp_prompt. These are registered immediately and made available via prompts/list and prompts/get.

property config: MCPConfig

Get the plugin configuration.

property registry: Registry

Get the central registry.

property discovered_tools: dict[str, BaseRouteHandler]

Get discovered MCP tools.

property discovered_resources: dict[str, BaseRouteHandler]

Get discovered MCP resources.

property discovered_prompts: dict[str, PromptRegistration]

Get discovered MCP prompts.

on_cli_init(cli)

Configure CLI commands for MCP operations.

Return type:

None

on_app_init(app_config)

Initialize the MCP integration when the Litestar app starts.

Return type:

AppConfig

on_startup(app)

Perform discovery after app is fully initialized and routes are built.

Return type:

None

Registry

class litestar_mcp.registry.Registry

Bases: object

Central registry for MCP tools, resources, and prompts.

This class decouples metadata storage and discovery from the route handlers themselves, avoiding issues with __slots__ or object mutation.

Note

Tools and resources are stored as bare BaseRouteHandler values because every entry has a single underlying handler. Prompts use PromptRegistration instead — a prompt may originate from either a standalone @mcp_prompt callable or a route handler, so the dataclass carries the fn | handler union plus the per-prompt metadata (title, description, arguments, icons) that can't live on a bare callable.

__init__()

Initialize the registry.

set_sse_manager(manager)

Set the SSE manager for notifications.

Return type:

None

property sse_manager: SSEManager

Return the configured SSE manager.

property tools: dict[str, BaseRouteHandler]

Get registered tools.

property resources: dict[str, BaseRouteHandler]

Get registered resources.

register_tool(name, handler)

Register a tool.

Parameters:

• name (str) -- The tool name.

• handler (BaseRouteHandler) -- The route handler.

Return type:

None

register_resource(name, handler)

Register a resource.

Parameters:

• name (str) -- The resource name.

• handler (BaseRouteHandler) -- The route handler.

Return type:

None

property templates: dict[str, ResourceTemplate]

Get registered resource templates, keyed by resource name.

register_resource_template(name, handler, template)

Register an RFC 6570 Level 1 URI template for a resource.

Parameters:

• name (str) -- The resource name (same key as register_resource).

• handler (BaseRouteHandler) -- The route handler bound to the template.

• template (str) -- The URI template string. Validated at registration; invalid templates raise ValueError.

Return type:

None

property prompts: dict[str, PromptRegistration]

Get registered prompts.

register_prompt(name, fn, *, title=None, description=None, arguments=None, icons=None)

Register a standalone prompt function.

Parameters:

• name (str) -- Unique prompt identifier.

• fn (Callable[..., Any]) -- The callable to invoke on prompts/get.

• title (Optional[str]) -- Optional human-readable display name.

• description (Optional[str]) -- Optional description. Falls back to fn.__doc__.

• arguments (Optional[list[dict[str, Any]]]) -- Explicit argument definitions. When None, derived from the function signature.

• icons (Optional[list[dict[str, Any]]]) -- Optional list of icon objects for UI display.

Return type:

None

register_prompt_handler(name, handler, *, title=None, description=None, arguments=None, icons=None)

Register a route-handler-based prompt.

Storage only — runtime dispatch and the messages-passthrough vs. normalize-on-return decision live in litestar_mcp.routes.handle_prompts_get(). This function captures the handler reference plus any explicit overrides so the registry can render prompts/list entries without executing the handler.

Parameters:

• name (str) -- Unique prompt identifier.

• handler (BaseRouteHandler) -- The Litestar route handler.

• title (Optional[str]) -- Optional human-readable display name.

• description (Optional[str]) -- Optional description.

• arguments (Optional[list[dict[str, Any]]]) -- Explicit argument definitions. When None, arguments are introspected from the handler's parsed handler signature at prompts/list render time (DI- and framework-injected parameters filtered out). Pass [] to advertise no arguments explicitly.

• icons (Optional[list[dict[str, Any]]]) -- Optional list of icon objects for UI display.

Return type:

None

async publish_notification(method, params, session_id=None)

Publish a JSON-RPC 2.0 notification to connected clients.

Parameters:

• method (str) -- The notification method (e.g., 'notifications/resources/updated').

• params (dict[str, Any]) -- The notification parameters.

• session_id (Optional[str]) -- Optional session to target; when omitted the notification fans out to every attached session.

Return type:

None

async notify_resource_updated(uri)

Notify clients that a resource has been updated.

Parameters:

uri (str) -- The URI of the updated resource.

Return type:

None

async notify_tools_list_changed()

Notify clients that the tool list has changed.

Return type:

None

async notify_prompts_list_changed()

Notify clients that the prompt list has changed.

Return type:

None

SSEManager

class litestar_mcp.sse.SSEManager

Bases: object

Manages Streamable HTTP SSE connections and message delivery.

The manager keeps one _StreamState per open stream and a side index from session_id to the set of stream ids currently attached to that session, so notifications can be fanned out to every stream belonging to a given MCP session.

__init__(*, max_streams=10000, max_idle_seconds=3600.0)

Initialize the SSE manager.

Parameters:

• max_streams (int) -- Hard cap on concurrent open streams. Excess raises StreamLimitExceeded.

• max_idle_seconds (float) -- Idle window (seconds) after which a stream is eligible for lazy pruning on the next open_stream.

async open_stream(session_id=None, last_event_id=None)

Open a new stream (or resume from last_event_id).

Parameters:

• session_id (Optional[str]) -- Optional session id to bind this stream to. When provided, the manager updates the session→streams index so publish() can fan out to all streams belonging to the session.

• last_event_id (Optional[str]) -- Optional Last-Event-ID value. On a match, pending messages from the existing stream's history are replayed before normal delivery resumes.

Return type:

tuple[str, AsyncGenerator[SSEMessage, None]]

Returns:

A (stream_id, async_generator) pair.

disconnect(stream_id)

Explicitly remove a stream and its buffered state.

Return type:

None

async enqueue(stream_id, message)

Enqueue a raw JSON payload onto a single stream.

Return type:

None

async publish(message, session_id=None)

Publish a JSON payload to one or all sessions.

When session_id is provided the message fans out to every stream attached to that session; otherwise it fans out to every stream attached to any session.

Return type:

None

async replay_from(stream_id, last_event_id)

Return buffered messages after last_event_id for a stream.

Return type:

list[SSEMessage]

close_session_streams(session_id)

Close every stream attached to session_id. Returns closed ids.

Return type:

list[str]

SSEMessage

class litestar_mcp.sse.SSEMessage

Bases: object

Represents a single SSE message.

__init__(data, event='message', id=None)