Litestar MCP Plugin

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.

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=[...]).

Related MCP server: backend2mcp

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.

• Bring Your Own Auth — MCP inherits the app's Litestar authentication middleware, including litestar-security or a custom AbstractAuthenticationMiddleware.

• Optional Task Support — the MCP Tasks extension with Litestar Store records; applications coordinate execution across workers.

• Optional A2A 1.0 — official SDK models and handlers on native Litestar JSON-RPC/SSE routes, without required Starlette, FastAPI or Uvicorn dependencies.

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 modern Streamable HTTP server
    mcp.run(transport="streamable-http", 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

• litestar --app my_app:app mcp stdio - in-process stdio for desktop clients

• litestar mcp bridge - stdio proxy to a running Streamable HTTP server

Install litestar-mcp[a2a] to mount an official A2A SDK request handler and agent card independently of MCP. The adapter accepts A2A 1.0 JSON-RPC requests with A2A-Version: 1.0; it rejects missing/legacy versions and offers no 0.3 conversion, gRPC or REST binding. Applications supply authentication, durable stores, executor resource scopes, worker coordination and push delivery policy. See the A2A guide.

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.

MCP IDs must be strings or finite integer-valued numbers. Missing, null, boolean, container or fractional IDs fail before tool execution or stream allocation. Progress streams apply bounded backpressure; slow subscription consumers are completed and disconnected. Shared task Stores persist records but do not distribute task execution or local input/cancel queues.

Google ADK 2.8.0 with MCP SDK 1.29.1 still sends the initialize-era lifecycle and cannot consume this endpoint. ADK RemoteA2aAgent interoperability has not been verified; local ADK agents and the tested official A2A SDK client are separate integration paths. See the 0.14 migration guide for removed aliases and configuration.

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:

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.

litestar-security

Use litestar-security for policy evaluation and RFC 9728 protected-resource metadata. Pass route policy metadata with MCPConfig(route_opt={"auth": required("api-key")}); evaluator failures can emit the required WWW-Authenticate resource metadata.

In-process stdio

Run litestar --app my_app:app mcp stdio to serve the application in-process to a desktop MCP client. Use mcp bridge when the MCP server is already running over Streamable HTTP.

Stdio uses native Litestar.lifespan(). Its shutdown_timeout bounds request cleanup and shutdown after startup completes; application hooks own bounded, cancellation-safe startup rollback. MCPStdioContext.session remains an application session. Provide a verified identity for protected tasks; anonymous stdio requests have no owner ID.

Development

# Clone the repository
git clone https://github.com/cofin/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 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.

The conformance runner reports passed, waived and failed scenarios separately. Its pinned alpha validator has documented task-extension schema waivers in tools/ci/run_mcp_conformance.py; MCP_CONFORMANCE_STRICT=1 disables those waivers. A waived check is not a protocol pass.

License

MIT License. See LICENSE for details.

Contributing

Contributions welcome! Please see our contribution guide for details.