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.

```bash
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.

```python
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`:

```python
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`.

```python
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:

```python
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:

```python
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]`):

```python
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:

```python
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 `lead_agent.name` /
  `lead_agent.description` at construction time.
- **`SupervisorAgent` forbids `tool_config` on the lead agent.** The supervisor manages tools
  internally. Use `extra_tools` for any additional tools beyond team dispatch.
- **`GroundedAgent` presenter isolation is strict.** The presenter never sees chat history or the
  gatherer's transcript — only the curated data produced by the `ToolOutputCurator`. A chit-chat
  turn that calls no tools is answered by the gatherer alone (presenter is skipped).
- **`MCPToolProvider.create` is async.** It must be `await`ed before building the agent. Call
  `await provider.disconnect()` when done to close server connections.
- **`save_chat=True` by default.** Every agent persists both sides of each exchange unless
  explicitly set to `False`. Storage is scoped per `(user_id, session_id, agent_id)`. The
  `MAX_MESSAGE_PAIRS_PER_AGENT` config trims at save time, not at fetch time.
- **`ConversationMessage.content` is a list of dicts**, not a plain string. Text is at
  `content[0]["text"]` for most agents. Some agents may produce multi-block content (tool use,
  images). Do not assume `content` has a single element.
- **`DakeraRetriever` and `AmazonKnowledgeBasesRetriever` are not import-guarded** — they are
  always exported from `agent_squad.retrievers`. If the underlying SDK (`dakera`, `boto3`) is not
  installed, the import will fail at runtime when you try to construct them.

## 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`, `TimestampedMessage`
- **Tests** — `python/src/tests/` — pytest; run from `python/` with `make test`.
- **Optional extras** — `python/setup.cfg` — `[options.extras_require]`.