Metadata-Version: 2.4
Name: litestar-mcp
Version: 0.6.0
Summary: Litestar plugin for Model Context Protocol (MCP) integration
Project-URL: Changelog, https://docs.litestar-mcp.litestar.dev/latest/changelog
Project-URL: Discord, https://discord.gg/litestar
Project-URL: Documentation, https://docs.litestar-mcp.litestar.dev/latest/
Project-URL: Funding, https://github.com/sponsors/litestar-org
Project-URL: Homepage, https://docs.litestar-mcp.litestar.dev/latest/
Project-URL: Issue, https://github.com/litestar-org/litestar-mcp/issues/
Project-URL: Source, https://github.com/litestar-org/litestar-mcp
Author-email: Cody Fincher <cody@litestar.dev>
Maintainer-email: Litestar Developers <hello@litestar.dev>, Cody Fincher <cody@litestar.dev>
License: MIT
License-File: LICENSE
Keywords: ai,litestar,mcp,model-context-protocol,plugin
Classifier: Development Status :: 4 - Beta
Classifier: Environment :: Web Environment
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: System Administrators
Classifier: License :: OSI Approved :: MIT License
Classifier: Natural Language :: English
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet :: WWW/HTTP :: Dynamic Content
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.24.1
Requires-Dist: litestar[jwt]>=2.23.0
Requires-Dist: typing-extensions>=4.0.0
Description-Content-Type: text/markdown

# 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](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"` 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

```bash
pip install litestar-mcp
# or
uv add litestar-mcp
```

### Basic Usage

```python
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

```python
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:**

```python
@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:**

```python
@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 metadata card
- `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`:

```python
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

```python
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:

```python
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`:

```python
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:

```python
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

```bash
# 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](LICENSE) for details.

## Contributing

Contributions welcome! Please see our [contribution guide](docs/contribution-guide.rst) for details.
