Auto-activate for litestar imports, litestar_granian, litestar_saq, litestar_email, litestar_mcp, litestar_vite imports, Litestar app configuration. Litestar ASGI web framework: route handlers, Guards, middleware, msgspec DTOs, OpenAPI, ecosystem plugins. Produces Litestar ASGI route handlers, middleware, guards, DTOs, and plugin configurations. Use when: building Litestar APIs, defining routes/controllers, configuring plugins (Granian, SAQ, Email, MCP, Vite), or working with Litestar dependency injection. Not for FastAPI, Django, or Flask -- Litestar has its own patterns.
From flownpx claudepluginhub cofin/flow --plugin flowThis skill uses the workspace's default tool permissions.
references/deployment.mdreferences/di.mdreferences/domains.mdreferences/dto.mdreferences/guards.mdreferences/litestar-email.mdreferences/litestar-granian.mdreferences/litestar-mcp.mdreferences/litestar-saq.mdreferences/middleware.mdreferences/pagination.mdreferences/plugins.mdreferences/routing.mdreferences/vite.mdreferences/websockets.mdSearches, retrieves, and installs Agent Skills from prompts.chat registry using MCP tools like search_skills and get_skill. Activates for finding skills, browsing catalogs, or extending Claude.
Searches prompts.chat for AI prompt templates by keyword or category, retrieves by ID with variable handling, and improves prompts via AI. Use for discovering or enhancing prompts.
Enables AI agents to execute x402 payments with per-task budgets, spending controls, and non-custodial wallets via MCP tools. Use when agents pay for APIs, services, or other agents.
Litestar is a high-performance Python ASGI web framework with built-in OpenAPI support, dependency injection, and first-class msgspec integration.
T | None (not Optional[T])from __future__ import annotationsfrom litestar import get, post, Controller
from litestar.di import Provide
@get("/items/{item_id:int}")
async def get_item(item_id: int) -> Item:
return await fetch_item(item_id)
class ItemController(Controller):
path = "/items"
dependencies = {"service": Provide(get_service)}
@get("/")
async def list_items(self, service: ItemService) -> list[Item]:
return await service.list_all()
from litestar.di import Provide
async def get_db_session(state: State) -> AsyncSession:
return state.db_session
app = Litestar(
route_handlers=[...],
dependencies={"session": Provide(get_db_session)},
)
from litestar.connection import ASGIConnection
from litestar.handlers import BaseRouteHandler
async def requires_auth(
connection: ASGIConnection,
_: BaseRouteHandler,
) -> None:
if not connection.user:
raise PermissionDeniedException("Authentication required")
@get(guards=[requires_auth])
async def protected_route() -> dict: ...
import msgspec
from litestar.dto import MsgspecDTO, DTOConfig
class UserStruct(msgspec.Struct):
id: int
name: str
class UserDTO(MsgspecDTO[UserStruct]):
config = DTOConfig(exclude={"password_hash"})
from litestar import Litestar
from litestar_granian import GranianPlugin
app = Litestar(plugins=[GranianPlugin()])
# Zero-config: replaces uvicorn CLI, adds `litestar run` via Granian
# See references/litestar-granian.md for logging configuration
from litestar import Litestar
from litestar_saq import SAQPlugin, SAQConfig, QueueConfig
saq = SAQPlugin(config=SAQConfig(
use_server_lifespan=True,
queue_configs=[QueueConfig(name="default", dsn="redis://localhost:6379/0")],
))
app = Litestar(plugins=[saq])
# Inject queues: async def handler(queues: TaskQueues) -> ...
# CLI: litestar workers run
# See references/litestar-saq.md for full config
from litestar import Litestar
from litestar_email import EmailPlugin, EmailConfig, SMTPConfig
app = Litestar(plugins=[EmailPlugin(config=EmailConfig(
backend=SMTPConfig(host="smtp.example.com", port=587, use_tls=True),
from_email="noreply@example.com",
))])
# Inject: async def handler(email_service: EmailService) -> ...
# Backends: SMTP, SendGrid, Resend, Mailgun, InMemory (testing)
# See references/litestar-email.md
from litestar import Litestar, get
from litestar_mcp import LitestarMCP, MCPConfig
@get("/users", name="list_users")
async def get_users() -> list[dict]: ...
app = Litestar(
route_handlers=[get_users],
plugins=[LitestarMCP(MCPConfig(name="My API"))],
)
# Mark routes: @mcp_tool("tool_name") or @mcp_resource("resource_name")
# Endpoint: POST /mcp/ (JSON-RPC 2.0)
# See references/litestar-mcp.md
from litestar import Litestar
from pathlib import Path
from litestar_vite import VitePlugin, ViteConfig, PathConfig
app = Litestar(plugins=[VitePlugin(config=ViteConfig(
dev_mode=True,
paths=PathConfig(root=Path(__file__).parent),
))])
# Modes: spa, template, ssr, framework
# CLI: litestar assets init|install|build|serve|generate-types
# See references/vite.md
<workflow>
Create msgspec Structs for request/response shapes. Use DTOConfig to control field inclusion, exclusion, and renaming for OpenAPI alignment.
Use @get, @post, @put, @delete decorators or group related endpoints in a Controller class. Inject dependencies via function parameters.
Apply guards for auth/authz at the route, controller, or app level. Use AbstractMiddleware for cross-cutting concerns (logging, timing, CORS).
Wire controllers into Router instances, register routers with the Litestar app, and configure plugins (Vite, SQLAlchemy, Dishka DI).
Confirm OpenAPI schema at /schema reflects the correct DTOs. Run the app with litestar run --reload and test endpoints.
Provide() dependencies. Never instantiate services inside handlers.from __future__ import annotations -- breaks Litestar's runtime type introspection for DI, DTOs, and OpenAPI.async def handlers and await database/HTTP calls. Sync handlers block the event loop.Controller class with shared path and dependencies.Before delivering Litestar code, verify:
from __future__ import annotations anywhereasync defTask: CRUD endpoint with guard, DTO, and dependency injection.
import msgspec
from litestar import Controller, get, post, delete
from litestar.connection import ASGIConnection
from litestar.di import Provide
from litestar.dto import MsgspecDTO, DTOConfig
from litestar.exceptions import PermissionDeniedException, NotFoundException
from litestar.handlers import BaseRouteHandler
# --- Guard ---
async def requires_auth(
connection: ASGIConnection,
_: BaseRouteHandler,
) -> None:
if not connection.user:
raise PermissionDeniedException("Authentication required")
# --- Domain model ---
class Task(msgspec.Struct):
id: int
title: str
done: bool = False
owner_id: int | None = None
class TaskCreate(msgspec.Struct):
title: str
# --- DTOs ---
class TaskReadDTO(MsgspecDTO[Task]):
config = DTOConfig(exclude={"owner_id"})
class TaskCreateDTO(MsgspecDTO[TaskCreate]):
pass
# --- Service (injected) ---
class TaskService:
async def list_for_user(self, user_id: int) -> list[Task]:
...
async def create(self, data: TaskCreate, owner_id: int) -> Task:
...
async def delete(self, task_id: int) -> None:
...
async def get_task_service() -> TaskService:
return TaskService()
# --- Controller ---
class TaskController(Controller):
path = "/tasks"
guards = [requires_auth]
dependencies = {"service": Provide(get_task_service)}
return_dto = TaskReadDTO
@get("/")
async def list_tasks(self, service: TaskService, request: "Request") -> list[Task]:
return await service.list_for_user(request.user.id)
@post("/", dto=TaskCreateDTO)
async def create_task(
self, data: TaskCreate, service: TaskService, request: "Request"
) -> Task:
return await service.create(data, owner_id=request.user.id)
@delete("/{task_id:int}", return_dto=None)
async def delete_task(self, task_id: int, service: TaskService) -> None:
await service.delete(task_id)
</example>
For detailed guides and configuration examples, refer to the following documents in references/: