Skip to main content
Glama
cofin
by cofin

Litestar MCP Plugin

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 Templatesmcp_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.

A
license - permissive license
-
quality - not tested
C
maintenance

Maintenance

Maintainers
Response time
5wRelease cycle
8Releases (12mo)
Issues opened vs closed

Resources

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/cofin/litestar-mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server