agent-squad-python

name

agent-squad-python

description

Use when building or modifying a Python app that uses the agent-squad Python package — async multi-agent orchestration for Python 3.11+: orchestrator, agents (BedrockLLMAgent, AnthropicAgent, OpenAIAgent, SupervisorAgent, GroundedAgent, ChainAgent, and more), classifier routing (Bedrock, Anthropic, OpenAI), storage (in-memory, DynamoDB, SQL/Turso), retrievers (Amazon KB, Dakera), tools (AgentTools, MCPToolProvider), and custom implementations.

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.

Agent	Extra needed	Notes
BedrockLLMAgent	aws	Bedrock Converse API; supports streaming, tools, retriever
AmazonBedrockAgent	aws	Bedrock Agents runtime (managed agents with KB/action groups)
BedrockInlineAgent	aws	Bedrock inline agents — code interpretation, KB, and tools inline
BedrockFlowsAgent	aws	Bedrock Flows — runs a preconfigured flow
BedrockTranslatorAgent	aws	Bedrock translation agent
LambdaAgent	aws	Invokes an AWS Lambda function
LexBotAgent	aws	Routes to an Amazon Lex bot
ComprehendFilterAgent	aws	Comprehend PII/toxicity filter before passing to another agent
ChainAgent	aws	Sequential pipeline — each agent's output feeds the next
AnthropicAgent	anthropic	Anthropic Messages API; supports streaming and tools
OpenAIAgent	openai	OpenAI Chat Completions API; supports streaming and tools
StrandsAgent	strands-agents	Strands Agents integration
SupervisorAgent	aws or anthropic	Lead agent coordinates a team via tools; always available in __init__.py but requires a compatible lead_agent
GroundedAgent	same as gatherer/presenter	Two-LLM anti-hallucination pattern; always importable

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

Classifier	Extra needed
BedrockClassifier	aws
AnthropicClassifier	anthropic
OpenAIClassifier	openai

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

Storage	Extra needed	Notes
InMemoryChatStorage	none	Default; not persistent
DynamoDbChatStorage	aws	DynamoDB-backed; production default for AWS deployments
SqlChatStorage	sql	libSQL/Turso-backed
SummarizingChatStorage	none	Wraps any storage; compresses history via a user-supplied async callable when history exceeds trigger_at pairs

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

Retriever	Extra needed	Notes
AmazonKnowledgeBasesRetriever	aws	Amazon Bedrock Knowledge Bases
DakeraRetriever	dakera	Self-hosted Dakera memory server

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

Custom implementations

Subclass the abstract base and pass your instance where the built-in goes. Source paths are under python/src/agent_squad/.

Seam	Base class	Source file
Agent	Agent	agents/agent.py
Classifier	Classifier	classifiers/classifier.py
Storage	ChatStorage	storage/chat_storage.py
Retriever	Retriever	retrievers/retriever.py
Tool curator	ToolOutputCurator	agents/grounded_agent.py
Presenter prompt	PresenterPrompt	agents/grounded_agent.py

Minimal custom agent:

from typing import Optional, AsyncIterable, Union
from agent_squad.agents import Agent, AgentOptions
from agent_squad.types import ConversationMessage, ParticipantRole

class MyAgent(Agent):
    def __init__(self, options: AgentOptions):
        super().__init__(options)

    async def process_request(
        self,
        input_text: str,
        user_id: str,
        session_id: str,
        chat_history: list[ConversationMessage],
        additional_params: Optional[dict] = None,
    ) -> Union[ConversationMessage, AsyncIterable]:
        return ConversationMessage(
            role=ParticipantRole.ASSISTANT.value,
            content=[{"text": f"Echo: {input_text}"}],
        )

For a streaming custom agent, also override is_streaming_enabled to return True and yield AgentStreamResponse objects (set final_message on the last one).

Gotchas

• All public methods are async. route_request, process_request, classify, storage methods, and retriever methods are all coroutines. You must await them inside an async context. Use asyncio.run(main()) at the top level.
• Optional imports at module level, not inside methods. The framework guards all optional integrations with try/except ImportError at the top of each module. If you copy this pattern for your own extensions, keep the guard at module level — never inside __init__ or a method.
• Classifier is required. Unlike the Swift version, the Python AgentSquad raises at construction time if no classifier can be resolved (no boto3 installed and no classifier passed). Always pass a classifier explicitly when boto3 is not available.
• response.streaming and response.output type are coupled. When streaming=True, output is an async generator; when False, it is a ConversationMessage. Always branch on response.streaming before consuming output.
• stream_response=False by default. Even if the agent itself streams internally, the orchestrator will drain the stream and return a single ConversationMessage unless you pass stream_response=True to route_request.
• Agent id is derived from name via generate_key_from_name: lowercased, spaces replaced with hyphens, special characters stripped. agent.id is the storage key — two agents with names that normalise to the same string will collide. Pick distinct names.
• AgentOptions is a dataclass; new fields must have defaults. When subclassing (e.g. BedrockLLMAgentOptions), add new fields with defaults so existing construction call-sites keep working.
• SupervisorAgent name and description come from the lead agent. Whatever you set on SupervisorAgentOptions.name / .description is overwritten by / at construction time.

Go deeper

• Prose and recipes — the Starlight docs under docs/src/content/docs/ (run with npm run dev from docs/): get-started/, agents/, classifiers/, storage/, retrievers/, tools/.
• Exact signatures — python/src/agent_squad/:
  • orchestrator.py — AgentSquad, route_request, classify_request, agent_process_request
  • agents/agent.py — Agent, AgentOptions, AgentCallbacks, AgentResponse, AgentStreamResponse
  • agents/grounded_agent.py — GroundedAgent, GroundedAgentOptions, ToolOutputCurator, DataBlockCurator, PerToolCurator, PresenterPrompt, CapturedToolResult
  • agents/supervisor_agent.py — SupervisorAgent, SupervisorAgentOptions
  • agents/chain_agent.py — ChainAgent, ChainAgentOptions
  • classifiers/classifier.py — Classifier, ClassifierResult, ClassifierCallbacks
  • storage/chat_storage.py — ChatStorage
  • retrievers/retriever.py — Retriever
  • utils/tool.py — AgentTools, AgentTool, AgentToolCallbacks, AgentToolResult
  • tools/mcp_tool_provider.py — MCPToolProvider, MCPServerConfig
  • types/types.py — ConversationMessage, ParticipantRole, AgentSquadConfig,