/litestar-ai

name: litestar-ai
description: Auto-activate for Google ADK, LlmAgent, Runner, SQLSpecSessionService, google-genai, AgentRuntime, SpecTree, DynamicWorkflow, FunctionTool, or SSE agent chats. Not for offline ML training.

Litestar AI & Custom Agent Engineering

Build production AI agents, multi-agent specialist hierarchies, multi-step workflows, and structured LLM services in Litestar using either **Google ADK (`google-adk` + `sqlspec.extensions.adk`)** or a **native `google-genai` + `msgspec` agent runtime**.

Code Style Rules

• **PEP 604 unions only**: `T | None`, never `Optional[T]`. Use built-in `list` and `dict` generics.
• **Async all I/O**: every model call, tool execution, session read/write, memory search, and cache lookup must be `async`.
• **`msgspec.Struct` wire contracts**: define request, response, SSE event, and runtime state payloads as `msgspec.Struct` (unless the project already standardizes on Pydantic).
• **Short-lived database transactions**: never hold open database connections across LLM generation turns; bind request-scoped services via an invocation-scoped `RunContextRegistry` token and acquire DB sessions inside individual tool calls or persistence steps.
• **Google-style docstrings**: document every agent factory, tool function, runtime service, and controller handler. Tool docstrings and parameter types define the model-visible tool schema.

Quick Reference

Architecture Selection (Match-Your-Stack)

| Pattern | Primary Signals | Best For | | --- | --- | --- | | **Google ADK + SQLSpec** (`google.adk` + `sqlspec.extensions.adk`) | `LlmAgent`, `App`, `Runner`, `SQLSpecSessionService`, `SQLSpecMemoryService`, `FunctionTool`, `SkillToolset` | Multi-turn conversational agents using ADK's built-in `sub_agents` delegation, `ContextCacheConfig`, `EventsCompactionConfig`, and `BasePlugin` callbacks backed by SQLSpec stores | | **Native `google-genai` + `msgspec` Runtime** (`google.genai` + `msgspec` + `sqlspec`) | `AgentRuntime`, `AgentSpec`, `SpecTree`, `DynamicWorkflow`, `WorkflowEngine`, `declare`, `CacheRegistry`, `CompactionPolicy` | Explicit control over DAG workflows (`WorkflowNode`), parallel tool execution (`asyncio.gather`) with live SSE events, optimistic session locking (`expected_version`), and custom prompt slot fencing | | **Structured-Output Domain Service** (`google.genai` + `GenerateContentConfig`) | `response_mime_type="application/json"`, `response_schema`, `embed_content`, `tenacity.retry` | Single-shot or batch classification, entity extraction, and embedding pipelines without multi-turn conversation state | | **MCP Server Surface** (`litestar-mcp`) | `LitestarMCP`, `MCPConfig`, `mcp_tool=`, `mcp_resource=` | Exposing Litestar route handlers or domain services to external MCP clients (use `litestar-mcp`) |

Session State Scope Prefixes

| Prefix | Lifetime | Typical Keys | Persistence Behavior | | --- | --- | --- | --- | | `app:*` | Application-wide | `app:catalog_version`, `app:feature_flags` | Shared across all users and sessions for `app_name` | | `user:*` | Principal-scoped | `user:preferred_region`, `user:output_format` | Shared across sessions belonging to the same `(workspace_id, user_id)` | | *(unprefixed)* | Session-scoped | `usage.total_tokens`, `compaction`, `workspace_id` | Persisted on the individual conversation session row | | `temp:*` | Single turn | `temp:run_context_token`, `temp:dynamic_instruction`, `temp:tool_calls` | Invocation-local only; stripped before database persistence |

Public SSE Event Contract

| `event` | Payload Shape (`data`) | Purpose | | --- | --- | --- | | `session` | `{"session_id": "...", "compacted_through_seq": 12}` | Emitted first so the client can bind the active session ID | | `tool_call` | `{"id": "...", "name": "...", "author": "...", "args": {...}}` | Model requested a tool invocation | | `tool_start` | `{"id": "...", "name": "...", "author": "...", "args": {...}}` | Parallel tool runner or workflow node started execution | | `tool_end` | `{"id": "...", "name": "...", "author": "...", "status": "completed" \| "error", "result_preview": "..."}` | Tool finished with bounded JSON preview | | `delta` | `{"text": "..."}` | Incremental assistant Markdown text chunk (thoughts excluded) | | `ping` | `{}` | Keep-alive frame emitted on idle intervals before terminal frame | | `complete` | `{"prompt_tokens": ..., "output_tokens": ..., "total_tokens": ..., "cached_tokens": ..., "finish_reason": "STOP"}` | Terminal success frame with aggregated turn token usage | | `error` | `{"code": "...", "message": "..."}` | Terminal error frame (`session_turn_limit`, `llm_calls_exhausted`, `session_conflict`, `stream_error`) |

<workflow>

1. Choose the Agent Runtime Pattern

1. Inspect `pyproject.toml` and existing agent modules. 2. If `google-adk` is installed or `LlmAgent` / `Runner` is present, follow the **Google ADK + SQLSpec** pattern in [adk-and-serving.md](references/adk-and-serving.md). 3. If `google-genai` is used directly with `msgspec.Struct` events, `SpecTree`, or `DynamicWorkflow`, follow the **Native `google-genai` Runtime** pattern in [custom-agents-and-tools.md](references/custom-agents-and-tools.md). 4. If the task is non-conversational classification, extraction, or embedding, implement a **Structured-Output Domain Service** with `response_mime_type="application/json"` and `tenacity` backoff on HTTP 429.

2. Design Agent Hierarchies & Prompt Isolation

1. **Coordinator + Specialist Tree**:

• In Google ADK, compose `LlmAgent(name="coordinator", sub_agents=[...], static_instruction=..., instruction=_dynamic_instruction)`.
• In the native runtime, build a `SpecTree(root=coordinator, specialists={...}, declarations={...})` which automatically injects an enum-constrained `transfer_to_agent` tool into every agent in the tree.

2. **Static vs Dynamic Instructions**:

• Keep `static_instruction` strictly immutable across turns and tenants so Vertex AI / Gemini context c