Litestar MCP

# uv (recommended)
uv add litestar-mcp

# pip
pip install litestar-mcp

# pdm
pdm add litestar-mcp

# poetry
poetry add litestar-mcp

uv add "litestar-mcp[a2a,queues,google-genai,opentelemetry]"

from litestar import Litestar, delete, get, post
from litestar.openapi.config import OpenAPIConfig
from litestar_mcp import LitestarMCP, MCPConfig


@get("/users", mcp_tool="list_users")
async def get_users(limit: int = 20) -> list[dict[str, object]]:
    """List active users in the workspace."""
    return [{"id": 1, "name": "Alice", "limit": limit}]


@post(
    "/reports/analyze",
    mcp_tool=True,
    mcp_annotations={"title": "Analyze Dataset", "idempotentHint": True},
)
async def analyze_data(dataset: str, threshold: float = 0.95) -> dict[str, object]:
    """Run statistical analysis over a named dataset."""
    return {"dataset": dataset, "threshold": threshold, "status": "ok"}


@delete("/users/{user_id:int}", mcp_tool="remove_user")
async def delete_user(user_id: int) -> dict[str, object]:
    """Delete a user account by ID."""
    return {"deleted": user_id}


@get("/config", mcp_resource="app_config")
async def get_app_config() -> dict[str, object]:
    """Return read-only application configuration metadata."""
    return {"environment": "production", "version": "2.0.0"}


app = Litestar(
    route_handlers=[get_users, analyze_data, delete_user, get_app_config],
    plugins=[LitestarMCP(MCPConfig(name="Workspace Service"))],
    openapi_config=OpenAPIConfig(title="Workspace Service", version="2.0.0"),
)

from litestar_mcp import MCP

mcp = MCP("math-service", instructions="Provides deterministic arithmetic tools.")


@mcp.tool()
def add(a: int, b: int) -> int:
    """Return the sum of two integers."""
    return a + b


@mcp.resource("config://limits", name="service_limits")
def limits() -> dict[str, int]:
    """Return service rate and payload limits."""
    return {"max_batch": 100}


app = mcp.app

if __name__ == "__main__":
    mcp.run(port=8000)

# MCP 2026-07-28 stateless discovery
curl -sX 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":{}}}}'

# MCP 2025-06-18 initialize handshake
curl -sX POST http://127.0.0.1:8000/mcp \
  -H 'Content-Type: application/json' \
  -H 'MCP-Protocol-Version: 2025-06-18' \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"curl","version":"1.0"}}}'

# Serve the application over stdin/stdout in-process with full Litestar lifespan
litestar --app my_app:app mcp stdio

# Bridge local stdio to a remote Streamable HTTP deployment
litestar mcp bridge --base-url https://api.example.com --endpoint /mcp

# Export static {"tools": [...]} JSON schemas without booting an HTTP server
litestar --app my_app:app mcp export-tool-spec --indent 2

from litestar import Litestar, post
from litestar.stores.memory import MemoryStore
from litestar.stores.registry import StoreRegistry
from litestar_queues import QueueConfig, QueuePlugin

from litestar_mcp import LitestarMCP, MCPConfig, MCPTaskConfig
from litestar_mcp.agent import DynamicWorkflow, WorkflowContext
from litestar_mcp.queues import DurableWorkflowEngine, QueueTaskExecutor


@post("/backfill", mcp_tool="run_backfill", mcp_task=True)
async def run_backfill(batch_size: int = 500) -> dict[str, object]:
    """Execute a background backfill task on a litestar-queues worker."""
    return {"processed": batch_size, "status": "completed"}


async def extract_step(ctx: WorkflowContext) -> dict[str, object]:
    """Extract source records for the workflow run."""
    return {"rows": 250}


pipeline = DynamicWorkflow("etl_pipeline")
pipeline.add_node("extract", handler=extract_step)

queue_plugin = QueuePlugin(QueueConfig(backend="memory"))
task_executor = QueueTaskExecutor(queue="mcp", timeout=3600)
workflow_engine = DurableWorkflowEngine([pipeline])

app = Litestar(
    route_handlers=[run_backfill],
    plugins=[
        queue_plugin,
        workflow_engine,
        LitestarMCP(
            MCPConfig(
                tasks=MCPTaskConfig(executor=task_executor),
            )
        ),
    ],
    stores=StoreRegistry(
        stores={
            "litestar_mcp_tasks": MemoryStore(),
            "litestar_mcp_workflows": MemoryStore(),
        }
    ),
)

from litestar import Litestar
from litestar_mcp.agent import (
    Agent,
    AgentChatController,
    AgentRuntime,
    GoogleGenAIClient,
    MemorySessionStore,
    TurnBudget,
)
from litestar_mcp.core.tools import tool


@tool
def lookup_order(order_id: str) -> dict[str, str]:
    """Look up order shipment status by ID."""
    return {"order_id": order_id, "status": "in_transit"}


support_agent = Agent(
    name="support",
    description="Answers customer order inquiries.",
    instructions="Use lookup_order to verify shipment status before replying.",
    tools=[lookup_order],
)

runtime = AgentRuntime(
    support_agent,
    default_model=GoogleGenAIClient(),
    sessions=MemorySessionStore(),
    budget=TurnBudget(max_turns=6, max_tool_calls=8, timeout_seconds=30.0),
)

app = Litestar(route_handlers=[AgentChatController], plugins=[runtime])

from litestar import Litestar
from litestar_mcp.agent import agent_to_a2a

a2a_plugin = agent_to_a2a(
    support_agent,
    path="/a2a",
    base_url="https://agents.example.com",
    runtime=runtime,
)

app = Litestar(plugins=[runtime, a2a_plugin])

uv run docs/examples/notes/sqlspec/no_auth.py
uv run docs/examples/notes/sqlspec/jwt_auth_dishka.py
uv run docs/examples/notes/sqlspec/google_iap.py
uv run docs/examples/notes/advanced_alchemy/no_auth.py

git clone https://github.com/cofin/litestar-mcp.git
cd litestar-mcp

# Create the virtual environment with all extras and dev groups
make install

# Run linters, type checkers (mypy + pyright), slotscheck, and unit/example tests
make lint
make test

# Run the pinned MCP 2026-07-28 conformance suite (in-process and queue executors)
make conformance
make conformance-queue

# Run the documentation verification pipeline (docstring AST scan, CLI check, build, artifact verification)
make docs-check