agent-squad-python

# 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`