--- title: "Litestar MCP · Litestar Plugins" canonical_url: "https://litestar.dev/plugins/litestar-mcp" last_updated: "2026-08-16T03:52:43.085Z" meta: description: "Litestar plugin for Model Context Protocol (MCP) integration" "og:description": "Litestar plugin for Model Context Protocol (MCP) integration" "og:title": "Litestar MCP · Litestar Plugins" "twitter:description": "Litestar plugin for Model Context Protocol (MCP) integration" "twitter:title": "Litestar MCP · Litestar Plugins" --- # ![Litestar MCP](https://raw.githubusercontent.com/litestar-org/plugin-registry/main/icons/litestar-mcp.svg) Litestar MCP Litestar plugin for Model Context Protocol (MCP) integration [**22K downloads**](https://pypistats.org/packages/litestar-mcp)• [**12 stars**](https://github.com/cofin/litestar-mcp)• [**v0.12.0**](https://github.com/cofin/litestar-mcp/releases) [![cofin](https://github.com/cofin.png?size=40)**cofin**](https://github.com/cofin) A lightweight plugin that integrates Litestar web applications with the Model Context Protocol (MCP) by exposing marked routes as MCP tools, resources, and prompts over MCP Streamable HTTP and JSON-RPC. [![PyPI - Version](https://img.shields.io/pypi/v/litestar-mcp)](https://pypi.org/project/litestar-mcp/) [![Python Version](https://img.shields.io/pypi/pyversions/litestar-mcp)](https://pypi.org/project/litestar-mcp/) [![License](https://img.shields.io/badge/license-MIT-blue.svg)](https://opensource.org/licenses/MIT) ## Overview This plugin automatically discovers Litestar routes marked for MCP and exposes them through an MCP-native transport surface. Pass `**mcp_tool="name"**`, `**mcp_resource="name"**`, or `**mcp_prompt="name"**` straight through to `**@get**` / `**@post**` / etc. — Litestar funnels unknown kwargs into `**handler.opt**`, so no second decorator or `**opt={...}**` wrapper is needed. Standalone prompt callables that are not bound to an HTTP route can be registered through `**LitestarMCP(prompts=[...])**`. ## Features - **Protocol-Native Transport** — MCP Streamable HTTP with JSON-RPC requests and SSE streams. - **Three MCP Primitives** — tools, resources, *and* prompts, with `**prompts/list**` and `**prompts/get**` driven by the same handler discovery as the rest of the surface. - **Simple Route Marking** — pass `**mcp_tool**` / `**mcp_resource**` / `**mcp_prompt**` kwargs straight through to Litestar's route decorators, or register standalone prompts via `**LitestarMCP(prompts=[...])**`. - **RFC 6570 URI Templates** — `**mcp_resource_template="app://…/{var}"**` dispatches concrete URIs to handlers with extracted vars. - **First-Class Descriptions** — structured `**mcp_description**`, `**mcp_agent_instructions**`, `**mcp_when_to_use**`, `**mcp_returns**` kwargs. - **Type Safe** — full type hints with dataclasses; `**msgspec**`-powered tool-argument validation. - **Automatic Discovery** — routes are discovered at app initialization. - **OpenAPI Integration** — server info derived from OpenAPI config. - **OIDC Auth Baked In** — bearer-token validation via `**MCPAuthBackend**` or a composable `**create_oidc_validator()**` factory; injectable `**JWKSCache**` protocol for shared document caches. - **Optional Task Support** — experimental in-memory MCP task lifecycle endpoints. ## Quick Start ### Installation ``` pip install litestar-mcp # or uv add litestar-mcp ``` ### Basic Usage ``` from litestar import Litestar, get, post from litestar.openapi.config import OpenAPIConfig from litestar_mcp import LitestarMCP # Mark routes for MCP exposure using the opt attribute @get("/users", mcp_tool="list_users") async def get_users() -> list[dict]: """List all users in the system.""" return [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}] @post("/analyze", mcp_tool="analyze_data") async def analyze_data(data: dict) -> dict: """Analyze the provided data and return insights.""" return {"result": f"Analyzed {len(data)} items"} @get("/config", mcp_resource="app_config") async def get_app_config() -> dict: """Get the current application configuration.""" return {"debug": True, "version": "1.0.0"} # Add the MCP plugin to your Litestar app app = Litestar( route_handlers=[get_users, analyze_data, get_app_config], plugins=[LitestarMCP()], openapi_config=OpenAPIConfig(title="My API", version="1.0.0"), ) ``` ### Standalone Application (Alternative) If you are building a standalone MCP server, you can use the `**MCP**` class which provides a simplified declarative API and programmatically boots the server using the standard Litestar CLI: ``` from litestar_mcp import MCP # 1. Initialize the application mcp = MCP("my-mcp-server", instructions="Exposes utility tools.") # 2. Register tools, resources, or prompts using decorators @mcp.tool() def add(a: int, b: int) -> int: """Calculate the sum of two integers.""" return a + b # 3. Expose the app globally so that the CLI can discover it app = mcp.app if __name__ == "__main__": # 4. Boot the server using Server-Sent Events (SSE) mcp.run(port=8000) ``` The standalone decorators accept Litestar route-handler keyword arguments such as `**dependencies**`, `**guards**`, `**response_headers**`, `**responses**`, `**summary**`, `**tags**`, DTO options, hooks, and arbitrary extra kwargs stored in `**handler.opt**`. The `**name**` keyword names the MCP primitive; use `**route_name**` to set Litestar's route-handler name separately. ### With Configuration ``` from litestar_mcp import LitestarMCP, MCPConfig config = MCPConfig( base_path="/api/mcp", # Change the base path name="Custom Server Name", # Override server name include_in_schema=True, # Include MCP routes in OpenAPI schema ) app = Litestar( route_handlers=[get_users, analyze_data, get_app_config], plugins=[LitestarMCP(config)], openapi_config=OpenAPIConfig(title="My API", version="1.0.0"), ) ``` ## Resources vs Tools: When to Use Each ### Use Resources (`**mcp_resource**`) for - **Read-only data** that AI models need to reference - **Static or semi-static information** like documentation, schemas, configurations - **Data that doesn't require parameters** to retrieve - **Reference material** that AI models should "know about" **Examples:** ``` @get("/schema", mcp_resource="database_schema") async def get_schema() -> dict: """Database schema information.""" return {"tables": ["users", "orders"], "relationships": [...]} @get("/docs", mcp_resource="api_docs") async def get_documentation() -> dict: """API documentation and usage examples.""" return {"endpoints": [...], "examples": [...]} ``` ### Use Tools (`**mcp_tool**`) for - **Actions that perform operations** or mutations - **Dynamic queries** that need input parameters - **Operations that change state** in your application - **Computations or data processing** tasks **Examples:** ``` @post("/users", mcp_tool="create_user") async def create_user(user_data: dict) -> dict: """Create a new user account.""" # Perform user creation logic return {"id": 123, "created": True} @get("/search", mcp_tool="search_data") async def search(query: str, limit: int = 10) -> dict: """Search through application data.""" # Perform search with parameters return {"results": [...], "total": 42} ``` ## How It Works 1. **Route Discovery**: At app initialization, the plugin scans all route handlers for the `**opt**` attribute 2. **Automatic Exposure**: Routes marked with `**mcp_tool**` or `**mcp_resource**` are automatically exposed 3. **MCP Transport**: The plugin adds a Streamable HTTP MCP endpoint under the configured base path (default `**/mcp**`) 4. **Server Info**: Server name and version are derived from your OpenAPI configuration ## MCP Endpoints Once configured, your application exposes these MCP-compatible endpoints: - `**POST /mcp**` - stateless MCP `**2026-07-28**` JSON-RPC and subscription streams - `**GET /.well-known/agent-card.json**` - Agent metadata card - `**GET /.well-known/oauth-protected-resource**` - OAuth protected resource metadata when auth is configured Use `**server/discover**` instead of an initialize handshake: ``` curl -X POST http://127.0.0.1:8000/mcp \ -H 'Content-Type: application/json' \ -H 'MCP-Protocol-Version: 2026-07-28' \ -H 'Mcp-Method: server/discover' \ -d '{"jsonrpc":"2.0","id":1,"method":"server/discover","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{},"io.modelcontextprotocol/clientInfo":{"name":"curl","version":"1"}}}}' ``` Every request is independent. There are no protocol sessions, sticky-routing headers, GET/DELETE transport handlers, or SSE replay. **Built-in Resources:** - `**litestar://openapi**` - Your application's OpenAPI schema (always available via `**resources/read**`) ## Configuration Configure the plugin using `**MCPConfig**`: ``` from litestar_mcp import MCPConfig config = MCPConfig() ``` **Configuration Options:** | **Option** | **Type** | **Default** | **Description** | | --- | --- | --- | --- | | `**base_path**` | `**str**` | `**"/mcp"**` | Base path for the MCP Streamable HTTP endpoint | | `**include_in_schema**` | `**bool**` | `**False**` | Whether to include MCP routes in OpenAPI schema | | `**name**` | `**str \| None**` | `**None**` | Override server name. If None, uses OpenAPI title | | `**guards**` | `**list[Any] \| None**` | `**None**` | Litestar guards applied to the MCP router | | `**allowed_origins**` | `**list[str] \| None**` | `**None**` | Exact additional Origins; present Origins must be same-origin or allowlisted | | `**include_operations**` | `**list[str] \| None**` | `**None**` | Only expose matching operation names | | `**exclude_operations**` | `**list[str] \| None**` | `**None**` | Exclude matching operation names | | `**include_tags**` | `**list[str] \| None**` | `**None**` | Only expose routes with matching OpenAPI tags | | `**exclude_tags**` | `**list[str] \| None**` | `**None**` | Exclude routes with matching OpenAPI tags | | `**auth**` | `**MCPAuthConfig \| None**` | `**None**` | Metadata for `**/.well-known/oauth-protected-resource**` discovery | | `**tasks**` | `**bool \| MCPTaskConfig**` | `**False**` | Enable the `**io.modelcontextprotocol/tasks**` extension | | `**cache_ttl_ms**` | `**int**` | `**0**` | Conservative cache lifetime for discovery/list/resource results | | `**cache_scope**` | `**"private" \| "public"**` | `**"private"**` | Cache sharing policy | | `**subscription_max_streams**` | `**int**` | `**10000**` | Maximum concurrent `**subscriptions/listen**` streams | | `**subscription_keepalive_seconds**` | `**float**` | `**15.0**` | Subscription keepalive interval | | `**subscription_channels**` | `**ChannelsPlugin \| None**` | `**None**` | Optional cross-worker notification fan-out | | `**before_tool_call**` | `**BeforeToolCallHook \| None**` | `**None**` | Observe each `**tools/call**` before dispatch | | `**after_tool_call**` | `**AfterToolCallHook \| None**` | `**None**` | Observe each `**tools/call**` result, exception, and duration | ## Complete Example ``` from litestar import Litestar, get, post, delete from litestar.openapi.config import OpenAPIConfig from litestar_mcp import LitestarMCP, MCPConfig # Resources - read-only reference data @get("/users/schema", mcp_resource="user_schema") async def get_user_schema() -> dict: """User data model schema.""" return { "type": "object", "properties": { "id": {"type": "integer"}, "name": {"type": "string"}, "email": {"type": "string"} } } @get("/api/info", mcp_resource="api_info") async def get_api_info() -> dict: """API capabilities and information.""" return { "version": "2.0.0", "features": ["user_management", "data_analysis"], "rate_limits": {"requests_per_minute": 1000} } # Tools - actionable operations @get("/users", mcp_tool="list_users") async def list_users(limit: int = 10) -> dict: """List users with optional limit.""" # Fetch users from database return {"users": [{"id": 1, "name": "Alice"}], "total": 1} @post("/users", mcp_tool="create_user") async def create_user(user_data: dict) -> dict: """Create a new user account.""" # Create user logic return {"id": 123, "created": True, "user": user_data} @post("/analyze", mcp_tool="analyze_dataset") async def analyze_dataset(config: dict) -> dict: """Analyze data with custom configuration.""" # Analysis logic return {"insights": [...], "metrics": {...}} # Regular routes (not exposed to MCP) @get("/health") async def health_check() -> dict: return {"status": "healthy"} # MCP configuration mcp_config = MCPConfig( name="User Management API", base_path="/mcp" ) # Create Litestar app app = Litestar( route_handlers=[ get_user_schema, get_api_info, # Resources list_users, create_user, analyze_dataset, # Tools health_check # Regular route ], plugins=[LitestarMCP(mcp_config)], openapi_config=OpenAPIConfig( title="User Management API", version="2.0.0" ), ) ``` ## Authentication Authentication is a **Litestar middleware** concern. Apps with an existing auth middleware get MCP authentication for free — `**request.user**` and `**request.auth**` are populated before tool handlers run. Three integration paths: ### Path A — Bring Your Own Middleware If your Litestar app already ships an `**AbstractAuthenticationMiddleware**` (or Litestar's built-in JWT backends), MCP inherits it automatically: ``` from litestar import Litestar from litestar.middleware import DefineMiddleware from litestar_mcp import LitestarMCP, MCPConfig app = Litestar( route_handlers=[...], plugins=[LitestarMCP(MCPConfig())], middleware=[DefineMiddleware(YourAuthMiddleware)], # MCP gets this for free ) ``` See `**docs/examples/notes/sqlspec/google_iap.py**` for a runnable example. ### Path B — Built-in MCPAuthBackend For OIDC workloads, install the built-in `**MCPAuthBackend**`: ``` from litestar import Litestar from litestar.middleware import DefineMiddleware from litestar_mcp import LitestarMCP, MCPAuthBackend, MCPConfig, OIDCProviderConfig from litestar_mcp.auth import MCPAuthConfig app = Litestar( route_handlers=[...], plugins=[LitestarMCP(MCPConfig(auth=MCPAuthConfig( issuer="https://company.okta.com", audience="api://mcp-tools", )))], middleware=[DefineMiddleware( MCPAuthBackend, providers=[OIDCProviderConfig( issuer="https://company.okta.com", audience="api://mcp-tools", )], user_resolver=lambda claims, app: MyUser(sub=claims["sub"]), )], ) ``` JWKS auto-discovery, caching, and `**clock_skew**` tolerance are built in. See `**docs/examples/notes/sqlspec/cloud_run_jwt.py**` for the full pattern. ### Path C — Composable OIDC Factory `**create_oidc_validator()**` returns an async callable for use as `**MCPAuthBackend(token_validator=...)**` or inside your own middleware: ``` from litestar_mcp import create_oidc_validator validator = create_oidc_validator( "https://cloud.google.com/iap", "/projects/PROJECT_NUMBER/global/backendServices/SERVICE_ID", algorithms=("ES256",), jwks_cache_ttl=1800, ) ``` ## Development ``` # Clone the repository git clone https://github.com/litestar-org/litestar-mcp.git cd litestar-mcp # Install with development dependencies make install # Run the complete Python quality gate make check-all # Run the pinned MCP 2026-07-28 conformance suite make conformance # Build strict documentation and validate examples make docs make validate-examples validate-uvx validate-pep723 # Run example uv run python docs/examples/hello_world/main.py ``` The conformance target owns Node.js `**24.18.1**` through `**NODE_VERSION**` in the Makefile. When [nodenv](https://github.com/nodenv/nodenv) is available, the target selects that version with `**NODENV_VERSION**`; otherwise it uses the active `**node**`/`**npm**` installation. A local `**.node-version**` is ignored so contributors can use nodenv without changing repository state. ## License MIT License. See [LICENSE](https://github.com/cofin/litestar-mcp/blob/main/LICENSE) for details. ## Contributing Contributions welcome! Please see our [contribution guide](https://github.com/cofin/litestar-mcp/blob/main/docs/contribution-guide.rst) for details.