Agent Squad Python Guide

agent-squad Python — assistant guide

Async-first, dependency-optional multi-agent orchestration framework (Python 3.11+). This is a guide and a map — not an API reference. Read exact signatures from the source (python/src/agent_squad/) and the docs site (docs/src/content/docs/); this file tells you what to use, when, and what to watch out for.

When to use what

• One assistant — a single Agent subclass; no orchestrator needed, call process_request directly.
• Several specialists — multiple agents + an AgentSquad orchestrator; the classifier routes each turn to the right agent automatically.
• Answers must not drift from data (prices, balances, live stock) — GroundedAgent: a gatherer LLM calls tools and sees raw results but never speaks to the user; an isolated presenter LLM writes the reply grounded only on curated facts.
• Fixed pipeline — ChainAgent: routes the output of one agent as the input to the next, sequentially.
• Team coordination — SupervisorAgent: a lead BedrockLLMAgent or AnthropicAgent delegates to a team of sub-agents via an internal tool loop, maintaining shared context. Can itself be registered in an AgentSquad.
• External tool servers — MCPToolProvider (requires agent-squad[mcp]) connects any number of MCP servers (stdio or SSE) and makes their tools available to any agent.

How to install

All third-party integrations are optional extras — never forced on users who don't need them.

pip install agent-squad                  # core only (no LLM runtime)
pip install "agent-squad[aws]"           # + boto3 — BedrockLLMAgent, BedrockClassifier, DynamoDbChatStorage, etc.
pip install "agent-squad[anthropic]"     # + anthropic SDK — AnthropicAgent, AnthropicClassifier
pip install "agent-squad[openai]"        # + openai SDK — OpenAIAgent, OpenAIClassifier
pip install "agent-squad[sql]"           # + libsql-client — SqlChatStorage (Turso/libSQL)
pip install "agent-squad[strands-agents]"# + strands-agents — StrandsAgent
pip install "agent-squad[dakera]"        # + dakera — DakeraRetriever
pip install "agent-squad[mcp]"           # + mcp — MCPToolProvider
pip install "agent-squad[all]"           # everything except strands-agents

How a turn works

AgentSquad.route_request is the one entry point worth memorising. It is a coroutine — you must await it.

import asyncio
from agent_squad.orchestrator import AgentSquad
from agent_squad.agents import BedrockLLMAgent, BedrockLLMAgentOptions
from agent_squad.classifiers import BedrockClassifier, BedrockClassifierOptions

orchestrator = AgentSquad(
    classifier=BedrockClassifier(BedrockClassifierOptions())
)
orchestrator.add_agent(BedrockLLMAgent(BedrockLLMAgentOptions(
    name="General Assistant",
    description="Handles general knowledge questions",
)))

async def main():
    response = await orchestrator.route_request(
        user_input="What is the capital of France?",
        user_id="user-123",
        session_id="session-abc",
    )

    if response.streaming:
        # response.output is an async generator of AgentStreamResponse
        async for chunk in response.output:
            if chunk.text:
                print(chunk.text, end="", flush=True)
            if chunk.final_message:
                pass  # full ConversationMessage — already persisted
    else:
        # response.output is a ConversationMessage
        print(response.output.content[0]["text"])

asyncio.run(main())

AgentResponse has three fields: metadata (AgentProcessingResult), output, and streaming (bool). Always branch on response.streaming — the type of output differs.

To stream back from route_request, pass stream_response=True:

response = await orchestrator.route_request(
    user_input="...",
    user_id="u1",
    session_id="s1",
    stream_response=True,
)

If no agent is selected and no default agent is configured, route_request returns an AgentResponse with the NO_SELECTED_AGENT_MESSAGE text rather than raising.

The pieces

AgentSquad (orchestrator)

from agent_squad.orchestrator import AgentSquad

The top-level object. Holds an agent registry, a classifier, and a ChatStorage.

from agent_squad.types import AgentSquadConfig

orchestrator = AgentSquad(
    options=AgentSquadConfig(
        LOG_CLASSIFIER_OUTPUT=True,
        MAX_MESSAGE_PAIRS_PER_AGENT=20,
        USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True,
    ),
    storage=my_storage,        # default: InMemoryChatStorage
    classifier=my_classifier,  # default: BedrockClassifier (if boto3 installed)
    default_agent=fallback,    # used when classifier returns no match
)
orchestrator.add_agent(agent)

AgentSquadConfig fields: LOG_AGENT_CHAT, LOG_CLASSIFIER_CHAT, LOG_CLASSIFIER_RAW_OUTPUT, LOG_CLASSIFIER_OUTPUT, LOG_EXECUTION_TIMES, MAX_RETRIES, USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED, NO_SELECTED_AGENT_MESSAGE, GENERAL_ROUTING_ERROR_MSG_MESSAGE, MAX_MESSAGE_PAIRS_PER_AGENT.

You can also call classify_request and agent_process_request separately if you need to inspect the routing decision before dispatching.

Agents

All agents require agent-squad[aws], [anthropic], or [openai] depending on the underlying SDK. The base Agent and AgentOptions plus SupervisorAgent and GroundedAgent are always available with the core install.

All agents extend Agent (agent_squad.agents.agent). The AgentOptions dataclass is the construction pattern — every concrete agent has a matching *Options dataclass that extends it:

from agent_squad.agents import BedrockLLMAgent, BedrockLLMAgentOptions
from agent_squad.utils import AgentTools, AgentTool

agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="My Agent",
    description="Handles X",              # shown to the classifier
    model_id="anthropic.claude-3-5-sonnet-20240620-v1:0",
    streaming=True,
    save_chat=True,                        # default True — persists history
    tool_config={"tool": my_tools},        # AgentTools instance
    retriever=my_retriever,
    LOG_AGENT_DEBUG_TRACE=False,
))

GroundedAgent takes a gatherer, presenter, tools, optional curator (ToolOutputCurator), and optional presenter_prompt (PresenterPrompt). The gatherer runs the tool loop; the presenter receives only the curated data and never sees the chat history or the gatherer's transcript.

SupervisorAgent takes a lead_agent (must be BedrockLLMAgent or AnthropicAgent), a team list, optional storage, and optional extra_tools. The lead agent must not have its own tool_config — tools are managed internally. Use extra_tools for additional tools beyond the team-dispatch tools.

ChainAgent (requires aws) takes an agents list. Each agent's text output becomes the next agent's input.

Classifiers

from agent_squad.classifiers import BedrockClassifier, AnthropicClassifier, OpenAIClassifier

The orchestrator defaults to BedrockClassifier if boto3 is installed and no classifier is provided. If boto3 is not installed and no classifier is passed, the orchestrator raises ValueError at construction time.

All classifiers extend Classifier (agent_squad.classifiers.classifier). You can override the routing prompt via set_system_prompt(template, variables) where {{AGENT_DESCRIPTIONS}} and {{HISTORY}} are the built-in template placeholders.

Storage

from agent_squad.storage import InMemoryChatStorage, DynamoDbChatStorage, SqlChatStorage

All storage classes extend ChatStorage (agent_squad.storage.chat_storage). Storage is keyed by (user_id, session_id, agent_id). MAX_MESSAGE_PAIRS_PER_AGENT (default 100) trims history at write time. save_chat=False on an agent disables history for that agent only.

Retrievers

from agent_squad.retrievers import AmazonKnowledgeBasesRetriever, DakeraRetriever

All retrievers extend Retriever (agent_squad.retrievers.retriever). A retriever attached to an agent augments its prompt with retrieved context before the LLM call.

DakeraRetriever reads DAKERA_API_KEY and DAKERA_URL from environment variables if not provided in DakeraRetrieverOptions.

Tools

AgentTools / AgentTool — the native tool system, always available:

from agent_squad.utils import AgentTools, AgentTool

def get_weather(city: str) -> str:
    """Get the weather for a city.
    :param city: The city name.
    """
    return f"Sunny in {city}"

tools = AgentTools(tools=[
    AgentTool(name="get_weather", func=get_weather)
])
# AgentTool auto-extracts properties from type hints and :param docstrings.

AgentTool wraps both sync and async functions transparently. Properties, descriptions, and required fields can be overridden explicitly. AgentTools.tool_handler processes tool call responses for Bedrock and Anthropic wire formats.

MCPToolProvider — drop-in AgentTools subclass for MCP servers (requires agent-squad[mcp]):

from agent_squad.tools import MCPToolProvider, MCPServerConfig

provider = await MCPToolProvider.create([
    MCPServerConfig(type="stdio", command="uvx", args=["my-mcp-server"]),
    MCPServerConfig(type="sse", url="http://localhost:3000/sse"),
])

agent = BedrockLLMAgent(BedrockLLMAgentOptions(
    name="MCP Agent",
    description="...",
    tool_config={"tool": provider},
))

# Clean up when done:
await provider.disconnect()

MCPToolProvider.create is an async factory — it connects to all servers and fetches tool definitions upfront so they are available synchronously when the agent builds its API request.

Callbacks

AgentCallbacks (on AgentOptions) hooks into the agent lifecycle:

• on_agent_start — returns a dict (tracking info) available to other callbacks via kwargs
• on_agent_end
• on_llm_start
• on_llm_new_token
• on_llm_end

AgentToolCallbacks (on AgentTools) hooks into tool execution:

• on_tool_start
• on_tool_end
• on_tool_error

ClassifierCallbacks hooks into the classifier:

• on_classifier_start
• on_classifier_stop