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 -- Plugin configuration. Defaults to MCPConfig().

• prompts -- 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: litestar_mcp.config.MCPConfig

Get the plugin configuration.

property registry: Registry

Get the central registry.

property task_store: MCPTaskStore | None

Get the task store.

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.

register_dynamic_handler(handler)

Register a dynamic route handler on the plugin.

This is typically used by the wrapper class to register decorated tools and resources.

Return type:

None

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

async on_shutdown(app)

Clean up resources on application shutdown.

Return type:

None

MCP

class litestar_mcp.app.MCP

Bases: object

A class that simplifies Model Context Protocol application setup.

This provides decorators and programmatic server execution for standalone applications.

The standalone application wrapper. Builds a Litestar app pre-configured with LitestarMCP, exposes @tool / @resource / @prompt decorators, and can serve over Streamable HTTP or stdio via MCP.run(). Pass a MCPStdioContext to run(transport="stdio", stdio_context=...) to seed the caller identity.

__init__(name, *, instructions=None, config=None, plugins=None, route_handlers=None, **kwargs)

property app: Litestar

Get the Litestar application instance.

This lazily instantiates the Litestar app upon first access, ensuring all dynamically registered handlers are captured.

tool(name=None, *, after_request=None, after_response=None, background=None, before_request=None, cache=False, cache_control=None, cache_key_builder=None, dependencies=None, dto=_EmptyEnum.EMPTY, etag=None, exception_handlers=None, guards=None, media_type=None, middleware=None, route_name=None, opt=None, request_class=None, response_class=None, response_cookies=None, response_headers=None, return_dto=_EmptyEnum.EMPTY, signature_namespace=None, status_code=None, sync_to_thread=None, content_encoding=None, content_media_type=None, deprecated=False, description=None, include_in_schema=_EmptyEnum.EMPTY, operation_class=<class 'litestar.openapi.spec.operation.Operation'>, operation_id=None, raises=None, response_description=None, responses=None, security=None, summary=None, tags=None, type_decoders=None, type_encoders=None, **kwargs)

Decorator to register a function as an MCP tool.

This dynamically wraps the function inside a Litestar route handler and registers it to the plugin. Additional keyword arguments are passed through to Litestar's post() route decorator.

resource(uri, *, name=None, mime_type=None, after_request=None, after_response=None, background=None, before_request=None, cache=False, cache_control=None, cache_key_builder=None, dependencies=None, dto=_EmptyEnum.EMPTY, etag=None, exception_handlers=None, guards=None, media_type=None, middleware=None, route_name=None, opt=None, request_class=None, response_class=None, response_cookies=None, response_headers=None, return_dto=_EmptyEnum.EMPTY, signature_namespace=None, status_code=None, sync_to_thread=None, content_encoding=None, content_media_type=None, deprecated=False, description=None, include_in_schema=_EmptyEnum.EMPTY, operation_class=<class 'litestar.openapi.spec.operation.Operation'>, operation_id=None, raises=None, response_description=None, responses=None, security=None, summary=None, tags=None, type_decoders=None, type_encoders=None, **kwargs)

Decorator to register a function as an MCP resource.

This dynamically wraps the function inside a Litestar route handler and registers it to the plugin. Additional keyword arguments are passed through to Litestar's get() route decorator.

prompt(name=None, *, after_request=None, after_response=None, background=None, before_request=None, cache=False, cache_control=None, cache_key_builder=None, dependencies=None, dto=_EmptyEnum.EMPTY, etag=None, exception_handlers=None, guards=None, media_type=None, middleware=None, route_name=None, opt=None, request_class=None, response_class=None, response_cookies=None, response_headers=None, return_dto=_EmptyEnum.EMPTY, signature_namespace=None, status_code=None, sync_to_thread=None, content_encoding=None, content_media_type=None, deprecated=False, description=None, include_in_schema=_EmptyEnum.EMPTY, operation_class=<class 'litestar.openapi.spec.operation.Operation'>, operation_id=None, raises=None, response_description=None, responses=None, security=None, summary=None, tags=None, type_decoders=None, type_encoders=None, **kwargs)

Decorator to register a function as an MCP prompt.

This dynamically wraps the function inside a Litestar route handler and registers it to the plugin. Additional keyword arguments are passed through to Litestar's get() route decorator.

run(transport='sse', **kwargs)

Run the MCP server using the specified transport.

Parameters:

• transport (Literal['sse', 'stdio']) -- The transport to use ("sse" or "stdio").

• **kwargs (typing.Any) -- Arguments passed to the runner.

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.

register_change_callback(callback)

Register a callback to be invoked when the registry changes.

Parameters:

callback -- The callback function.

unregister_change_callback(callback)

Unregister a change callback.

Parameters:

callback -- The callback function to remove.

set_subscription_manager(manager)

Set the subscription manager for notifications.

Return type:

None

property subscription_manager: SubscriptionManager

Return the configured subscription 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 -- Unique prompt identifier.

• fn -- The callable to invoke on prompts/get.

• title -- Optional human-readable display name.

• description -- Optional description. Falls back to fn.__doc__.

• arguments -- Explicit argument definitions. When None, derived from the function signature.

• icons -- Optional list of icon objects for UI display.

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, typing.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, typing.Any]]]) -- Optional list of icon objects for UI display.

Return type:

None

async publish_notification(method, params)

Publish a JSON-RPC 2.0 notification to connected clients.

Parameters:

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

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

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

SSEMessage