Agents#

Litestar MCP provides high-level agent primitives designed to eliminate repetitive boilerplate when building multi-agent systems, structured prompts, and domain skills. These abstractions integrate directly with Litestar's dependency injection, guards, route handlers, and streaming responses.

Architecture Overview#

The agent framework centers around three core layers:

  1. Controllers (:class:`~litestar_mcp.PromptController`, :class:`~litestar_mcp.SkillController`): Class-based boundaries mirroring Litestar's controller architecture to package prompts, tools, instructions, and authorization guards. Controller guards and DI apply directly to internal MCP route handlers.

  2. Agent Hierarchy (:class:`~litestar_mcp.agent.Agent`, :class:`~litestar_mcp.agent.AgentGroup`): Coordinator and specialist agent trees featuring automated, Literal-typed transfer_to_agent tools and non-mutating specialist isolation.

  3. Model & Runtime Layer (:class:`~litestar_mcp.agent.GoogleGenAIClient`, :class:`~litestar_mcp.agent.AgentRuntime`): Streaming model execution using gemini-3.8-flash, thought signature preservation, turn deadlines, owner-scoped session storage, and typed Server-Sent Events (SSE).

Prompt Controllers#

PromptController allows declaring class-based MCP prompt groups with class-level configurable attributes:

docs/examples/agent/coffee_shop_agent.py - CoffeePromptController#
class CoffeePromptController(PromptController):
    """Prompt templates for greeting customers and authoring seasonal menus."""

    prefix = "coffee"
    instructions = (
        "Persona: Warm, knowledgeable neighborhood specialty coffee barista.\n"
        "Tone: Welcoming, concise, and focused on quality craft brewing.\n"
    )

    @prompt(title="Welcome Greeting", description="Greet a customer and introduce daily specials.")
    def welcome_customer(self, customer_name: str = "Friend") -> str:
        """Generate a warm barista greeting."""
        return f"Welcome to the coffee bar, {customer_name}! What can we craft for you today?"

    @prompt(title="Tasting Notes", description="Generate tasting notes for a given roast.")
    def roast_notes(self, roast_name: str) -> list[PromptMessage]:
        """Produce structured tasting notes for single-origin coffees."""
        return [
            PromptMessage.user(
                f"Produce professional SCA-style tasting notes and brewing parameters for `{roast_name}`."
            )
        ]

Because the MCP specification limits prompt messages to the user and assistant roles, class-level instructions are prepended to the first user message rather than emitted as a forbidden system role. Controller-level guards and dependency injection providers are preserved across both prompts/list and prompts/get.

Skill Controllers#

A SkillController encapsulates domain tools, prompts, instructions, and guards into a reusable unit:

docs/examples/agent/coffee_shop_agent.py - BaristaSkillController#
class BaristaSkillController(SkillController):
    """Barista skill providing beverage menu lookup and order preparation tools."""

    name = "barista"
    description = "Specialized skills for querying menu items and preparing coffee orders"
    prefix = "barista"
    instructions = (
        "Barista Operational Grounding:\n"
        "- Suggest seasonal single-origin roasts when asked for recommendations.\n"
        "- Confirm milk alternatives (oat, almond, soy) before finalizing espresso drinks.\n"
        "- Never accept orders for items outside the published menu.\n"
    )

    def __init__(self, owner: Any = None) -> None:
        """Initialize the barista skill with catalog menu and empty order list."""
        super().__init__(owner)
        self.menu: dict[str, dict[str, Any]] = {
            "espresso": {"price": 3.50, "roast": "Ethiopian Yirgacheffe", "decaf_available": True},
            "flat_white": {"price": 4.75, "roast": "Guatemalan Antigua", "decaf_available": True},
            "cold_brew": {"price": 5.00, "roast": "Colombian Huila", "decaf_available": False},
        }
        self.orders: list[dict[str, Any]] = []

    @tool(description="Retrieve available coffee drinks and pricing.")
    def get_menu(self) -> dict[str, dict[str, Any]]:
        """Return the current coffee menu."""
        return self.menu

    @tool(description="Place a customized beverage order.")
    def place_order(self, beverage: str, milk: str = "whole", shots: int = 2) -> dict[str, Any]:
        """Record an order and return confirmation details."""
        bev = beverage.lower().replace(" ", "_")
        if bev not in self.menu:
            return {"status": "error", "message": f"'{beverage}' is not on the menu."}

        order_record = {
            "order_id": len(self.orders) + 1,
            "beverage": bev,
            "milk": milk,
            "shots": shots,
            "price": self.menu[bev]["price"],
            "status": "in_queue",
        }
        self.orders.append(order_record)
        return {"status": "success", "order": order_record}

Skill controllers compile each method into an internal Litestar route handler, retaining method guards, type-based context injection, and dependency parameters. Skill controllers can be mounted on LitestarMCP, attached to an Agent, or bridged to A2A AgentSkill cards.

Agents and Agent Groups#

An Agent bundles instructions, tools, skills, and model configuration. An AgentGroup coordinates multi-agent interactions, automatically synthesizing a non-mutating transfer_to_agent tool for the coordinator to route turns to specialists:

docs/examples/agent/coffee_shop_agent.py - setup_coffee_agent_group#
def setup_coffee_agent_group() -> AgentGroup:
    """Build and configure the coffee shop agent group with specialists."""
    model_client = GoogleGenAIClient(model="gemini-3.8-flash")

    barista_agent = Agent(
        name="barista",
        description="Prepares coffee drinks, looks up menu pricing, and manages customer orders.",
        instructions="You are the lead barista. Handle menu questions and beverage customizations.",
        skills=[BaristaSkillController()],
        model=model_client,
    )

    triage_coordinator = Agent(
        name="concierge",
        description="Greets customers and routes coffee questions to the barista specialist.",
        instructions="Greet customers warmly. Delegate ordering and coffee questions to the barista.",
        model=model_client,
    )

    return AgentGroup(
        coordinator=triage_coordinator,
        specialists=[barista_agent],
    )

Google GenAI Client#

GoogleGenAIClient provides first-party integration with google-genai, defaulting to gemini-3.8-flash and supporting Vertex AI endpoints, shared httpx2.AsyncClient transports, thinking budgets, and automatic thought signature round-tripping on follow-up function-calling turns:

docs/examples/agent/coffee_shop_agent.py - create_model_client#
def create_model_client() -> GoogleGenAIClient:
    """Instantiate GoogleGenAIClient with optional thinking budget."""
    return GoogleGenAIClient(
        model="gemini-3.8-flash",
        thinking_budget=2048,
        temperature=0.2,
    )

Sessions and Runtime#

AgentRuntime manages turn execution loops, tool invocations, and turn deadlines. History persistence is managed by pluggable SessionStore implementations:

  • MemorySessionStore: In-process LRU cache with TTL eviction and per-session concurrency locks.

  • LitestarStoreSessionStore: Distributed persistence backed by any Litestar Store (Redis, SQLite, file).

Sessions are strictly owner-scoped to prevent IDOR vulnerabilities. Anonymous sessions require server-generated IDs.

Realtime SSE Streaming#

Mounting AgentChatController provides HTTP endpoints and live Server-Sent Events (SSE) streaming with Last-Event-ID cursor recovery:

  • POST /agent/turns: Synchronous turn execution returning full aggregated output.

  • GET /agent/turns/{turn_id}/stream: Realtime SSE event stream emitting typed frames (thought, delta, tool_call, tool_result, complete) with keepalive pings.

Runnable Coffee Shop Example#

The repository provides a complete, runnable example in docs/examples/agent/coffee_shop_agent.py demonstrating a neighborhood barista agent group with custom tools, prompts, and streaming chat:

docs/examples/agent/coffee_shop_agent.py#
"""Executable reference example: Coffee Shop Barista Agent with Litestar MCP."""

from typing import Any

from litestar import Litestar
from litestar.di import Provide

from litestar_mcp import (
    LitestarMCP,
    PromptController,
    PromptMessage,
    SkillController,
    prompt,
    tool,
)
from litestar_mcp.agent import (
    Agent,
    AgentChatController,
    AgentGroup,
    AgentRuntime,
    GoogleGenAIClient,
)
from litestar_mcp.mcp.config import MCPConfig


class BaristaSkillController(SkillController):
    """Barista skill providing beverage menu lookup and order preparation tools."""

    name = "barista"
    description = "Specialized skills for querying menu items and preparing coffee orders"
    prefix = "barista"
    instructions = (
        "Barista Operational Grounding:\n"
        "- Suggest seasonal single-origin roasts when asked for recommendations.\n"
        "- Confirm milk alternatives (oat, almond, soy) before finalizing espresso drinks.\n"
        "- Never accept orders for items outside the published menu.\n"
    )

    def __init__(self, owner: Any = None) -> None:
        """Initialize the barista skill with catalog menu and empty order list."""
        super().__init__(owner)
        self.menu: dict[str, dict[str, Any]] = {
            "espresso": {"price": 3.50, "roast": "Ethiopian Yirgacheffe", "decaf_available": True},
            "flat_white": {"price": 4.75, "roast": "Guatemalan Antigua", "decaf_available": True},
            "cold_brew": {"price": 5.00, "roast": "Colombian Huila", "decaf_available": False},
        }
        self.orders: list[dict[str, Any]] = []

    @tool(description="Retrieve available coffee drinks and pricing.")
    def get_menu(self) -> dict[str, dict[str, Any]]:
        """Return the current coffee menu."""
        return self.menu

    @tool(description="Place a customized beverage order.")
    def place_order(self, beverage: str, milk: str = "whole", shots: int = 2) -> dict[str, Any]:
        """Record an order and return confirmation details."""
        bev = beverage.lower().replace(" ", "_")
        if bev not in self.menu:
            return {"status": "error", "message": f"'{beverage}' is not on the menu."}

        order_record = {
            "order_id": len(self.orders) + 1,
            "beverage": bev,
            "milk": milk,
            "shots": shots,
            "price": self.menu[bev]["price"],
            "status": "in_queue",
        }
        self.orders.append(order_record)
        return {"status": "success", "order": order_record}


class CoffeePromptController(PromptController):
    """Prompt templates for greeting customers and authoring seasonal menus."""

    prefix = "coffee"
    instructions = (
        "Persona: Warm, knowledgeable neighborhood specialty coffee barista.\n"
        "Tone: Welcoming, concise, and focused on quality craft brewing.\n"
    )

    @prompt(title="Welcome Greeting", description="Greet a customer and introduce daily specials.")
    def welcome_customer(self, customer_name: str = "Friend") -> str:
        """Generate a warm barista greeting."""
        return f"Welcome to the coffee bar, {customer_name}! What can we craft for you today?"

    @prompt(title="Tasting Notes", description="Generate tasting notes for a given roast.")
    def roast_notes(self, roast_name: str) -> list[PromptMessage]:
        """Produce structured tasting notes for single-origin coffees."""
        return [
            PromptMessage.user(
                f"Produce professional SCA-style tasting notes and brewing parameters for `{roast_name}`."
            )
        ]


def create_model_client() -> GoogleGenAIClient:
    """Instantiate GoogleGenAIClient with optional thinking budget."""
    return GoogleGenAIClient(
        model="gemini-3.8-flash",
        thinking_budget=2048,
        temperature=0.2,
    )


def setup_coffee_agent_group() -> AgentGroup:
    """Build and configure the coffee shop agent group with specialists."""
    model_client = GoogleGenAIClient(model="gemini-3.8-flash")

    barista_agent = Agent(
        name="barista",
        description="Prepares coffee drinks, looks up menu pricing, and manages customer orders.",
        instructions="You are the lead barista. Handle menu questions and beverage customizations.",
        skills=[BaristaSkillController()],
        model=model_client,
    )

    triage_coordinator = Agent(
        name="concierge",
        description="Greets customers and routes coffee questions to the barista specialist.",
        instructions="Greet customers warmly. Delegate ordering and coffee questions to the barista.",
        model=model_client,
    )

    return AgentGroup(
        coordinator=triage_coordinator,
        specialists=[barista_agent],
    )


coffee_shop_group = setup_coffee_agent_group()

agent_runtime = AgentRuntime(target=coffee_shop_group)

mcp_plugin = LitestarMCP(
    config=MCPConfig(
        prompt_controllers=[CoffeePromptController],
        skill_controllers=[BaristaSkillController],
    )
)

app = Litestar(
    route_handlers=[
        AgentChatController,
        CoffeePromptController,
        BaristaSkillController,
    ],
    plugins=[mcp_plugin],
    dependencies={"runtime": Provide(lambda: agent_runtime, sync_to_thread=False)},
)