Agent Groups & Handoffs¶
When a domain problem spans multiple areas of expertise, AgentGroup (litestar_mcp.agent.AgentGroup) coordinates a primary triage or routing Agent (coordinator) and one or more domain-focused Agent instances (specialists) within a single AgentRuntime turn budget and session transcript.
Coordinator-Specialist Topology¶
flowchart LR
User["User TurnRequest"]
Coord["Coordinator Agent"]
Transfer["transfer_to_agent Tool"]
SpecA["Barista Specialist"]
SpecB["Roastery Specialist"]
User --> Coord
Coord --> Transfer
Transfer --> SpecA
Transfer --> SpecB
Initialize an AgentGroup with a coordinator and a sequence of specialists:
from litestar_mcp.agent import Agent, AgentGroup
barista = 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],
)
concierge = 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.",
)
coffee_group = AgentGroup(
coordinator=concierge,
specialists=[barista],
)
At construction time, AgentGroup validates structural invariants and raises ImproperlyConfiguredException if:
- Any two agents in
[coordinator, *specialists]share the samename. - The
coordinatorinstance is also listed inspecialists.
Automatic transfer_to_agent Tool Synthesis¶
When specialists is non-empty, AgentGroup synthesizes a group-owned handoff tool named transfer_to_agent (litestar_mcp.agent.spec.TRANSFER_TOOL_NAME):
Literal-Constrained Target Schema:AgentGroupcollectsspecialist_names = tuple(self.specialists.keys())and dynamically annotatestarget_agent: Literal[specialist_names]along withreason: str = "". The resulting JSON Schema 2020-12 restrictstarget_agentto an exactenumof valid specialist names, preventing hallucinated handoff targets.- Non-Mutating
AgentSpecifications:AgentGroupnever mutatescoordinator.toolsorcoordinator.tool_set. Instead,AgentRuntimecallsgroup.tools_for(active_agent)on each iteration: - When
active_agent is group.coordinator,group.tools_for(coordinator)returns(*coordinator.tool_set, group._transfer_tool). - When
active_agentis a specialist,group.tools_for(specialist)returnsspecialist.tool_setunchanged. BecauseAgentspecifications remain immutable after__post_init__, the sameAgentinstance can be reused across multipleAgentGroupdefinitions or standaloneAgentRuntimeinstances without cross-contaminating tool catalogs. - Reserved Tool Name:
Agent.__post_init__reserves"transfer_to_agent"and raisesImproperlyConfiguredExceptionif anAgentdeclares a custom tool with that name.
Specialist Context Isolation & Handoff Lifecycle¶
During AgentRuntime._execute_turn_loop, a handoff from the coordinator to a specialist proceeds within the same turn and session lock:
| Step | Runtime Action | Emitted AgentStreamFrame |
|---|---|---|
| 1. Coordinator Turn | AgentRuntime invokes coordinator's resolved ModelClient with coordinator.combined_instructions and group.tools_for(coordinator). |
thought, delta, tool_call (tool_name="transfer_to_agent") |
| 2. Tool Execution | transfer_to_agent(target_agent="barista", reason="...") executes and appends a ToolResult ({"status": "transferred", "target_agent": "barista", "reason": "..."}) to messages. |
tool_result |
| 3. Transfer Resolution | _check_agent_transfer passes the ToolCall to group.resolve_transfer(call), switching active_agent to barista. |
agent_transfer (agent_name="barista", payload={"reason": "..."}) |
| 4. Specialist Turn | The next loop iteration resolves barista's ModelClient, barista.combined_instructions, and barista.tool_set (for example get_menu and place_order) over the shared messages transcript. |
thought, delta, tool_call, tool_result, complete (agent_name="barista") |
Key isolation properties:
- Instruction & Tool Scoping: Once control transfers to a specialist, the model prompt uses only that specialist's
combined_instructions(including itsSkillControllerrules) andtool_set. The coordinator's tools andtransfer_to_agentare not exposed to the specialist. - Per-Specialist Model Binding: Each specialist
Agentcan specify its ownmodel(for example a reasoning-heavy model on a specialist and a fast router model on the coordinator).AgentRuntime._resolve_model(active_agent)switches model clients automatically on handoff. - Shared
TurnBudget: Coordinator and specialist iterations share the turn'sTurnBudget(max_turns,max_tool_calls,timeout_seconds), preventing infinite transfer cycles. - Attribution in
AgentMessageandTurnResponse: EveryAgentMessagerecorded inSessionStoreand everyAgentStreamFrameis stamped withagent_name=active_agent.name, andTurnResponse.agent_namereports the specialist that completed the turn.
Complete Walkthrough: Coffee Shop Agent Group¶
The reference application in docs/examples/agent/coffee_shop_agent.py combines a SkillController (BaristaSkillController), a PromptController (CoffeePromptController), a GoogleGenAIClient configured with a thinking budget, and an AgentGroup (concierge coordinator handing off to the barista specialist) mounted alongside AgentChatController and LitestarMCP:
"""Executable reference example: Coffee Shop Barista Agent with Litestar MCP."""
from typing import Any
from litestar import Litestar
from litestar_mcp import (
LitestarMCP,
PromptController,
PromptMessage,
SkillController,
prompt,
tool,
)
from litestar_mcp.agent import (
Agent,
AgentChatController,
AgentGroup,
AgentRuntime,
GoogleGenAIClient,
)
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"
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
def get_menu(self) -> dict[str, dict[str, Any]]:
"""Retrieve available coffee drinks and pricing."""
return self.menu
@tool
def place_order(self, beverage: str, milk: str = "whole", shots: int = 2) -> dict[str, Any]:
"""Place a customized beverage order."""
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 a thinking budget and low temperature."""
return GoogleGenAIClient(thinking_budget=2048, temperature=0.2)
def setup_coffee_agent_group() -> AgentGroup:
"""Build and configure the coffee shop agent group with specialists."""
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],
)
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.",
)
return AgentGroup(
coordinator=triage_coordinator,
specialists=[barista_agent],
)
app = Litestar(
route_handlers=[AgentChatController, CoffeePromptController, BaristaSkillController],
plugins=[
AgentRuntime(setup_coffee_agent_group(), default_model=create_model_client()),
LitestarMCP(),
],
)
How the Pieces Connect¶
BaristaSkillController: Declaresname = "barista", operationalinstructions, and two@toolmethods (get_menuandplace_order). Attachingskills=[BaristaSkillController]tobarista_agentinjects both tools intobarista_agent.tool_setand appends the operational grounding rules tobarista_agent.combined_instructions. PassingBaristaSkillControllerinLitestar(route_handlers=[...])simultaneously exposes those tools to MCP clients on/mcp.CoffeePromptController: Exposescoffee/welcome_customerandcoffee/roast_notesover MCPprompts/listandprompts/getwith prepended persona instructions.setup_coffee_agent_group(): Wiresconciergeas thecoordinatorandbaristaas the specialist insideAgentGroup.AgentRuntime&AgentChatController:AgentRuntime(setup_coffee_agent_group(), default_model=create_model_client())registers theruntimedependency used byAgentChatController(/agent/chat,/agent/turns,/agent/sessions/{session_id}).