Litestar

A lightweight plugin that integrates Litestar web applications with the Model Context Protocol (MCP) by exposing marked routes as MCP tools and resources over MCP Streamable HTTP and JSON-RPC.

PyPI - Version Python Version License

Overview

This plugin automatically discovers Litestar routes marked for MCP and exposes them through an MCP-native transport surface. Pass mcp_tool="name" or mcp_resource="name" straight through to @get / @post / etc. — Litestar funnels unknown kwargs into handler.opt, so no second decorator or opt={...} wrapper is needed.

Features

• Protocol-Native Transport — MCP Streamable HTTP with JSON-RPC requests and SSE streams.
• Simple Route Marking — pass mcp_tool / mcp_resource kwargs straight through to Litestar's route decorators.
• 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"),
)

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:

• GET /mcp - Server-Sent Events stream when Accept: text/event-stream is provided
• POST /mcp - JSON-RPC endpoint for initialize, ping, tools/*, resources/*, and optional task methods
• GET /.well-known/mcp-server.json - MCP server manifest
• GET /.well-known/agent-card.json - Agent card metadata
• GET /.well-known/oauth-protected-resource - OAuth protected resource metadata when auth is configured

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	Restrict accepted Origin header values
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 experimental in-memory MCP task support

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
uv sync --all-extras --dev

# Run tests
make test

# Run example
uv run python docs/examples/hello_world/main.py

License

MIT License. See LICENSE for details.

Contributing

Contributions welcome! Please see our contribution guide for details.