8  Production Agent Platforms for Python

Batteries-Included Agents with Agno and MCP-Native Composition with mcp-agent

Abstract

PocketFlow proved that an agent framework is just a graph and a shared store — and that’s the right place to learn. Chapter 7 added Burr + Pydantic for typed state, immutable snapshots, and fork-from-state replay. But production agents need more than a state machine. They need MCP plumbing, multi-agent coordination, RAG, memory, and composable patterns — and assembling those from independent libraries is a sprint of integration work. This chapter covers two Python platforms that ship those layers as first-class primitives. Agno is a batteries-included agent operating system: four primitives (Agent, Team, Workflow, Knowledge) with built-in MCP, ten storage backends, and OpenTelemetry tracing. mcp-agent treats the Model Context Protocol as the substrate and ships Anthropic’s canonical agent patterns (Router, Parallel, Evaluator-Optimizer) as composable factories with Temporal for durable execution. We compare them along the axes that matter at production scale, then build a worked example in each.

8.1 When PocketFlow Stops Being Enough

Three pressures emerge between the toy and the deployment:

1. Multiple people touch the agent. A mutable shared dict is a contract anyone can break. Without declared reads/writes, any node can silently overwrite any other’s data — and the breakage shows up two iterations downstream, not where it happened.
2. A production run fails at 3 a.m. and you need to debug it. “What was shared["context"] when the model went off the rails?” The answer requires either a print statement you added in advance or a state snapshot you kept. PocketFlow has neither.
3. MCP servers proliferate. Once Claude Code, your codebase, your monitoring, and three vendor MCP servers are all tools your agent needs to reach, the per-server wiring (process lifecycle, auth, sampling, elicitation) becomes its own subsystem. Hand-rolling it is the same job over and over.

Chapter 7 addressed the first two pressures with Burr + Pydantic — typed state, declared reads/writes, and fork-from-state replay. This chapter addresses the third pressure and the broader platform question: once you need MCP integration, multi-agent coordination, RAG, memory, and canonical patterns, which framework ships them ready to go?

flowchart TD
    P["PocketFlow + Burr<br/>(Ch 7: learn + type)"] --> A["Agno<br/>batteries-included platform"]
    P --> M["mcp-agent<br/>MCP-native + canonical patterns"]
    A --> Q["Production deployment"]
    M --> Q
    style P fill:#fef3c7,stroke:#92400e,color:#92400e
    style A fill:#dcfce7,stroke:#166534,color:#166534
    style M fill:#f3e8ff,stroke:#6b21a8,color:#6b21a8
    style Q fill:#fce7f3,stroke:#9d174d,color:#9d174d

Two platforms, two centers of gravity — with Burr from Chapter 7 for comparison

The two frameworks aren’t drop-in substitutes. Agno is closer to an agent operating system; mcp-agent is a pattern library on top of MCP. The matrix in Section 8.5 makes the differences explicit.

Note

Chapter 7 covers Burr + Pydantic — the typed-FSM-with-replay pick. If your primary need is fork-from-state debugging and explicit state contracts, start there. This chapter assumes you’ve seen Burr and are now asking the platform question: “What ships MCP, RAG, multi-agent, and patterns as first-class primitives?”

8.2 Agno: Batteries-Included Agent Platform

Agno positions itself as an “agent operating system” — the AgentOS framing isn’t marketing, it’s an honest signal about scope. Where Burr gives you a state-machine library and asks you to assemble the rest, Agno ships the rest. MCP tooling, RAG with a Knowledge primitive, multi-agent Teams, fifteen storage backends for sessions and memory, OpenTelemetry, scheduling, RBAC, and chat interfaces for Slack and Telegram are all in the box.

The decision to use Agno is therefore less “is the state machine model right?” and more “do I want a framework with opinions about every layer, or do I want to assemble those layers myself?”

ImportantVersion: this section is agno 3.0.0

Agno breaks public API across major versions. Every signature and every code block below was checked against agno 3.0.0 by introspection, and the runnable ones were executed. Code written for 1.x or 2.x fails at import or at the first keyword argument — Section 8.2.10 has the symptom-to-fix table.

Install: pip install "agno[mcp]" anthropic sqlalchemy. The extras matter. sqlalchemy backs SqliteDb/PostgresDb, fastapi+uvicorn back AgentOS, and the [mcp] extra pins mcp<2 — installing bare mcp gets you 2.0.0, where McpError was renamed MCPError, and Agno reports that as ImportError: `mcp` not installed even though it is. Do not chase that as a missing package.

8.2.1 The four primitives

from agno.agent import Agent
from agno.team import Team
from agno.workflow import Workflow
from agno.knowledge import Knowledge

• Agent — a single autonomous unit with tools, instructions, and a model. The smallest piece. Agents can hold memory, knowledge, skills, and MCP tools.
• Team — a coordinated set of agents that collaborate on a task. A leader model routes between members.
• Workflow — step-based execution with conditional logic, loops, parallel branches, and routers. You define Steps; your Python sequences them.
• Knowledge — a searchable corpus that any agent can query. Agno calls it “Agentic RAG”: retrieval is exposed as a tool the model calls, not a preprocessing step you hand-orchestrate.

What separates them is who decides control flow:

Primitive	Who decides what runs next	Use when
Agent	the model, one turn at a time	a single capability with tools
Team	a leader model picks members	routing needs judgment
Workflow	your Python picks steps	the pipeline shape is known at design time
Knowledge	the model decides when to search	retrieval belongs in the loop, not before it

They nest: a Team holds Agents and other Teams, an Agent holds Knowledge, and a Workflow step executes an Agent, a Team, a nested Workflow, or a plain Python callable.

8.2.1.1 Agent: the building block

The simplest Agno agent:

from agno.agent import Agent
from agno.models.anthropic import Claude

1agent = Agent(
    name="assistant",
2    model=Claude(id="claude-sonnet-4-6"),
    instructions="You are a helpful assistant. Be concise.",
    markdown=True,
)

3output = agent.run("What is context engineering?")
print(output.content)

1

Agent is the unit of work. Every parameter is keyword-only — name, model, instructions, and tools are all constructor arguments, and there is no subclassing.

2

Model is a first-class parameter. Agno ships 48 provider modules under agno.models.*. Swap Claude(...) for OpenAIResponses(...) to switch. Pin the id: Claude() alone defaults to claude-sonnet-4-5-20250929 in 3.0.0 and that default moves between releases.

3

agent.run(input) returns a RunOutput. The first argument is positional and named input — 1.x called it message, and run(message=...) now raises TypeError.

Four call shapes, differing only in async and printing:

Call	Returns	Use
agent.run(...)	RunOutput	programmatic, sync
agent.arun(...)	awaitable RunOutput	programmatic, async — also takes background=True
agent.print_response(...)	None	terminal demo, renders markdown
agent.aprint_response(...)	None	terminal demo, async

RunOutput carries more than content: messages, metrics, tools, citations, references, reasoning_content, status, session_state, requirements (pending human approvals), and the replay lineage fields run_id, parent_run_id, forked_from_run_id, forked_from_message_index, last_checkpoint_at_message_index.

Add tools and the agent becomes capable:

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.websearch import WebSearchTools
from agno.tools.yfinance import YFinanceTools

analyst = Agent(
    name="market_analyst",
    model=Claude(id="claude-sonnet-4-6"),
1    tools=[
        WebSearchTools(),
        YFinanceTools(stock_price=True, company_info=True),
    ],
2    instructions=[
        "You are a financial analyst.",
        "Always cite your data sources.",
        "Use tables for numerical comparisons.",
    ],
3    fallback_models=[Claude(id="claude-haiku-4-5-20251001")],
    retries=2,
    exponential_backoff=True,
    markdown=True,
)

analyst.print_response("Compare AAPL and MSFT performance this quarter.")

1

Tools are plug-and-play. Agno ships 141 toolkit modules under agno.tools.* — web search, GitHub, Jira, Slack, SQL, DuckDB, pandas, shell, Docker, finance, arXiv, PubMed, and so on. MCP tools use the same interface — see Section 8.2.3.

2

instructions accepts a list. Rendered as bullet lines. The idiomatic form for multi-rule agents; a plain str and a Callable also work.

3

Failover and retry are parameters, not wrappers. fallback_models covers a provider outage; retries / delay_between_retries / exponential_backoff cover transient errors. Writing a retry decorator around run() duplicates code that already exists.

Writing your own tool is a decorated function, and the decorator is also where human approval lives:

from agno.tools import tool

1@tool(requires_confirmation=True)
def deploy(env: str) -> str:
2    """Deploy the current build to an environment."""
    return f"deployed {env}"

1

requires_confirmation=True pauses the run before the call executes and surfaces the pending call to your code. This is the tool-granularity approval gate; Section 8.2.2 covers the step-granularity one.

2

The docstring is the tool description sent to the model. Write it for the model. A vague docstring is a tool the model calls at the wrong time.

8.2.1.2 Knowledge: agentic RAG

Agno’s Knowledge primitive is the RAG story. You point it at a vector database and sources; the agent gets a search_knowledge tool automatically:

from agno.knowledge import Knowledge
from agno.knowledge.embedder.openai import OpenAIEmbedder
from agno.vectordb.pgvector import PgVector
from agno.vectordb.search import SearchType

1knowledge = Knowledge(
    vector_db=PgVector(
        db_url="postgresql+psycopg://ai:ai@localhost:5532/ai",
        table_name="product_docs",
2        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

3knowledge.insert_many(
    paths=["./docs/**/*.md"],
    urls=["https://example.com/api-reference.pdf"],
)

1

Knowledge wraps a vector DB. Twenty backends live under agno.vectordb.*: PgVector, LanceDB, Qdrant, Pinecone, Weaviate, Chroma, Milvus, MongoDB, OpenSearch, Redis, Cassandra, Couchbase, ClickHouse, and adapters for LangChain and LlamaIndex retrievers.

2

SearchType.hybrid combines keyword and semantic search. Also available: vector (pure embedding similarity) and keyword (BM25). Import from agno.vectordb.search.

3

insert_many() ingests sources — markdown files, PDFs, URLs — into the vector DB. The keyword overload takes paths= and urls= lists separately; a list of content dicts also works as the first positional argument. For a single item, use insert(path=...) or insert(url=...).

Watch the defaults across those two methods: insert defaults skip_if_exists=False (re-ingest), insert_many defaults skip_if_exists=True (skip). Both have async twins, ainsert and ainsert_many.

Before adding a parsing dependency, check the built-in readers — knowledge.pdf_reader, .markdown_reader, .csv_reader, .docx_reader, .excel_reader, .pptx_reader, .json_reader, .website_reader, .youtube_reader, .firecrawl_reader. Remote sources ship too: S3Content, GCSContent, SharePointContent, GitHubContent, AzureBlobContent from agno.knowledge.remote_content.

Attach it to an agent:

from agno.agent import Agent
from agno.models.anthropic import Claude

support_agent = Agent(
    name="support",
    model=Claude(id="claude-sonnet-4-6"),
1    knowledge=knowledge,
    instructions="Answer questions using the product documentation. "
                 "Always search knowledge before answering.",
)

support_agent.print_response("How do I configure SSO?")

1

Attaching knowledge is the whole setup. The agent now has a search_knowledge tool, because search_knowledge defaults to True on Agent. You pass search_knowledge=False to turn retrieval off, not True to turn it on — a common cargo-culted line that does nothing.

The model decides when to call the search tool. That is the “agentic” part: you are not writing a retrieve-then-generate pipeline. If you want one, retrieve yourself and pass the text in through dependencies=.

Knowledge also supports agentic filters — metadata-based filtering where the model decides which filters to apply:

# Insert with shared metadata across this batch:
knowledge.insert_many(
    paths=["sso-guide.md"],
1    metadata={"product": "enterprise", "year": "2025"},
)
knowledge.insert_many(
    paths=["basic-auth.md"],
    metadata={"product": "starter", "year": "2024"},
)

agent = Agent(
    knowledge=knowledge,
2    enable_agentic_knowledge_filters=True,
)

1

metadata= is a keyword arg applied to all items in that call, not a per-item dict. To attach different metadata to different files, make separate insert_many calls (one per metadata group), as shown.

2

Agentic filters let the model add WHERE clauses to the vector search based on the user’s question. “How do I configure SSO for the enterprise tier?” automatically filters on product=enterprise. For fixed filters, set knowledge_filters={...} on the agent or pass it per call to run(); both accept a dict or a list of FilterExpr from agno.filters.

8.2.1.3 Workflow: step-based orchestration

When you need sequential stages with conditional branching, Workflow provides a step engine. All step types are exported from agno.workflow directly in 3.0:

from agno.workflow import Workflow, Step, Steps, Condition, Loop, Parallel, Router
from agno.workflow.types import StepInput, StepOutput

researcher = Agent(name="Researcher", instructions="Research the topic.",
                   tools=[WebSearchTools()])
fact_checker = Agent(name="Fact Checker", instructions="Verify claims.")
writer = Agent(name="Writer", instructions="Write the final article.")

1def needs_fact_checking(step_input: StepInput) -> bool:
    """Check if the research output contains claims needing verification."""
    return "claim" in str(step_input.previous_step_content).lower()

workflow = Workflow(
    name="Article Pipeline",
    steps=[
2        Step(name="research", agent=researcher),
3        Condition(
            name="fact_check_gate",
            evaluator=needs_fact_checking,
            steps=[Step(name="fact_check", agent=fact_checker)],
        ),
4        Step(name="write", agent=writer),
    ],
)

workflow.print_response("Write an article about MCP adoption trends.")

1

Conditions are Python callables. The evaluator receives a StepInput and returns a boolean. previous_step_content holds the prior step’s output; previous_step_outputs is a dict keyed by step name, which is how you reach back past the immediately preceding step.

2

Steps wrap an executor. Exactly one of agent=, team=, workflow=, or executor= per step.

3

Condition gates a branch, with an optional else_steps=.

4

Steps execute sequentially. The writer sees the output of the researcher (and fact checker, if it ran).

The five composite step types:

Type	Key parameters	What it does
Condition(steps, evaluator, else_steps=None)	evaluator is a callable, awaitable, bool, or str	branch
Loop(steps, max_iterations=3, end_condition=None)	end_condition: Callable[[List[StepOutput]], bool]	bounded repeat
Parallel(*steps)	steps as varargs, not a list	fan-out / fan-in
Router(choices, selector)	selector: Callable[[StepInput], List[step]]	pick a branch
Steps(...)	—	a named sub-sequence

The escape hatch worth knowing about is executor=: a plain callable (StepInput) -> StepOutput, needing no model at all.

def render(step_input: StepInput) -> StepOutput:
    path = ffmpeg_render(step_input.previous_step_content)
1    return StepOutput(content=path, stop=False)

Step(name="render", executor=render)

1

StepOutput.stop=True halts the whole workflow from inside a step. success=False plus error= marks a failure without stopping, which pairs with Step(skip_on_failure=True).

Deterministic work belongs here. A step with a plain function cannot drift, cannot burn tokens, and cannot be talked out of its behavior by a prompt.

TipWorkflow vs. Team

Workflow gives you deterministic step-by-step execution with conditional branches. Team gives you model-driven coordination where the leader agent decides which members to delegate to and how to synthesize their responses. Use Workflow when the pipeline shape is known at design time; use Team when the routing decision requires judgment. They compose — a Workflow step can execute a whole Team.

8.2.2 Human review: pausing on the irreversible act

Any step can carry a HumanReview, and the run pauses rather than proceeding:

from agno.db.sqlite import SqliteDb
from agno.workflow import Workflow, Step
from agno.workflow.types import HumanReview, StepInput, StepOutput

1db = SqliteDb(db_file="publish.db")

workflow = Workflow(
    name="publish",
    db=db,
    steps=[
        Step(name="draft",
             executor=lambda si: StepOutput(content=f"draft: {si.input}")),
        Step(name="publish",
             executor=lambda si: StepOutput(
                 content=f"PUBLISHED {si.previous_step_content}"),
2             human_review=HumanReview(
                 requires_confirmation=True,
                 confirmation_message="Ship it?",
                 on_reject="skip",
                 timeout=3600,
                 on_timeout="cancel",
             )),
    ],
)

run = workflow.run(input="episode 12")
3print(run.is_paused, run.pause_kind, run.paused_step_name)

4for req in run.steps_requiring_confirmation:
    req.confirm()                          # or req.reject(feedback="runs long")

5run = workflow.continue_run(run)
print(run.content)

1

A db is required for HITL. The pause has to survive the process; with no db there is nowhere to park it.

2

HumanReview is a per-step parameter, not a separate node. Its full surface: requires_confirmation, requires_user_input (with a user_input_schema), requires_output_review, requires_iteration_review (fires once per Loop pass), plus on_reject (skip / cancel / else / retry), on_error (fail / skip / pause), max_retries, timeout, and on_timeout (cancel / skip / approve).

3

The paused run tells you where it stopped — is_paused, pause_kind, paused_step_name, paused_step_index.

4

Pending approvals are objects with methods. confirm(), reject(feedback=...), set_user_input(...), select(...), edit(...). The run exposes them by kind: steps_requiring_confirmation, steps_requiring_output_review, steps_requiring_user_input, steps_requiring_route.

5

continue_run(run) resumes from the pause. Already-executed steps do not re-run.

That example is executable and its output is real:

True PauseKind.STEP publish
PUBLISHED draft: episode 12

NoteTwo granularities of approval

@tool(requires_confirmation=True) gates a single tool call inside an agent’s turn. HumanReview on a Step gates a whole stage of a workflow. Toolkit(requires_confirmation_tools=[...]) does the tool-level version for a whole toolkit. Reach for one of the three before building an approvals table — AgentOS already provisions one.

8.2.3 MCP integration via MCPTools

Agno’s MCP story is first-class. The MCPTools class connects to any MCP server — the tools it exposes appear to the agent exactly like native Python tools:

import asyncio
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools

async def main():
1    mcp_tools = MCPTools(
        "npx -y @modelcontextprotocol/server-brave-search",
        env={"BRAVE_API_KEY": "${BRAVE_API_KEY}"},
        timeout_seconds=30,
2        include_tools=["brave_web_search"],
        tool_name_prefix="brave",
    )
3    await mcp_tools.connect()

    agent = Agent(
        name="search_agent",
        model=Claude(id="claude-sonnet-4-6"),
4        tools=[mcp_tools],
        instructions="Use web search to answer questions with citations.",
    )

    await agent.aprint_response("What are the latest MCP spec changes?")
5    await mcp_tools.close()

asyncio.run(main())

1

MCPTools wraps a single MCP server. command is the first positional parameter; env, timeout_seconds, and transport settings follow as keywords.

2

include_tools / exclude_tools allowlist the surface, and tool_name_prefix avoids name collisions when two servers export a search. A twelve-tool server dumped whole into an agent’s context is a common and avoidable cost.

3

connect() starts the server process. Agno handles the lifecycle — process supervision, stdio or streamable-HTTP transport, and graceful shutdown.

4

MCP tools drop into the same tools=[] list. The agent doesn’t distinguish between native tools and MCP tools.

5

close() tears down the server. In AgentOS mode (Section 8.2.8), lifecycle is automatic.

For remote MCP servers (streamable HTTP), the connection is simpler — no process management, just a URL:

mcp_tools = MCPTools(
    transport="streamable-http",
    url="https://docs.agno.com/mcp",
1    header_provider=lambda: {"Authorization": f"Bearer {current_token()}"},
)

1

header_provider is called per run, which is how you attach a rotating token to a remote server.

WarningMultiMCPTools is gone in 3.0

Agno 2.x had a MultiMCPTools([...], allow_partial_failure=True) class for fanning out across several servers. It was removed in 3.0 — agno.tools.mcp now exports exactly MCPTools, SSEClientParams, and StreamableHTTPClientParams. Pass several MCPTools instances and own the failure policy yourself:

servers = [
    MCPTools("npx -y @modelcontextprotocol/server-brave-search"),
    MCPTools("npx -y @modelcontextprotocol/server-filesystem /tmp"),
]
live = []
for server in servers:
    try:
        await server.connect()
        live.append(server)
    except Exception as exc:
        log.warning("MCP server unavailable: %s", exc)   # was allow_partial_failure

agent = Agent(model=Claude(id="claude-sonnet-4-6"), tools=live)

Six lines, and now the degraded state is logged instead of silent — which is what you wanted in production anyway.

8.2.4 Memory, sessions, and replay

Agno separates two kinds of persistence, both configured through one db parameter:

• Session storage — messages and runs within a conversation.
• Memory — persistent user facts and preferences that survive across sessions, extracted by a model from the conversation.

8.2.4.1 Session storage backends

The same agent can persist to SQLite locally and Postgres in production by swapping one line:

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.db.postgres import PostgresDb

# Local development
agent = Agent(
1    db=SqliteDb(db_file="dev.db"),
2    add_history_to_context=True,
3    num_history_runs=6,
)

# Production
agent = Agent(
4    db=PostgresDb(db_url="postgresql+psycopg://..."),
    add_history_to_context=True,
    num_history_runs=6,
)

1

SQLite for development. Zero config, single file.

2

add_history_to_context=True automatically includes previous messages in the prompt.

3

num_history_runs=6 limits how many prior turns are included — a context window budget control. num_history_messages and max_tool_calls_from_history are the finer knobs.

4

Postgres for production. Same API, different backend. Fifteen backends live under agno.db.*: SQLite, Postgres (sync and async), MySQL, MongoDB, Redis, Valkey, DynamoDB, Firestore, ClickHouse, SingleStore, SurrealDB, JSON files, GCS-backed JSON, and in-memory.

Every *Db also takes per-table name overrides — SqliteDb alone exposes twenty-five of them (session_table, runs_table, memory_table, metrics_table, eval_table, traces_table, spans_table, learnings_table, schedules_table, approvals_table, and the MCP OAuth set). Sharing one database across several systems needs a naming convention, not a wrapper class.

8.2.4.2 Persistent user memory

Memory goes beyond session history — it captures facts about users that persist across sessions:

from agno.agent import Agent
from agno.db.postgres import PostgresDb

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

agent = Agent(
    db=db,
1    enable_agentic_memory=True,
2    update_memory_on_run=True,
)

# First conversation
agent.run("My name is Sarah and I prefer dark mode.", user_id="sarah")

# Later conversation (possibly a different session)
3output = agent.run("What are my preferences?", user_id="sarah")
# "You prefer dark mode."

1

enable_agentic_memory=True gives the agent tools to create, update, and delete user memories.

2

update_memory_on_run=True automatically extracts memories from each conversation turn.

3

Memories persist across sessions. The agent retrieves Sarah’s preferences from the database, not from the current conversation history.

Shared memory across agents is not an API — it is the same database:

agent_1 = Agent(db=db, update_memory_on_run=True)
agent_2 = Agent(db=db, update_memory_on_run=True)

agent_1.run("My name is John Doe", user_id="john")
1agent_2.run("What is my name?", user_id="john")

1

Agent 2 reads Agent 1’s memories. Both point at the same db, and user_id is the scope key.

Beyond a single session, search_past_sessions=True with num_past_sessions_to_search does semantic recall across a user’s prior conversations, and enable_session_summaries=True keeps a rolling summary instead of raw turns.

8.2.4.3 Learning: memory with structure

New in 3.0, LearningMachine is a curation layer above raw memory with six independently switchable stores:

from agno.learn import LearningMachine

agent = Agent(
    db=db,
    learning=LearningMachine(
        db=db,
        user_profile=True,        # stable facts about the person
        user_memory=True,         # episodic preferences
        session_context=True,     # what this session is about
        entity_memory=True,       # facts about things mentioned
        learned_knowledge=True,   # generalizations promoted into Knowledge
        decision_log=True,        # why the agent chose what it chose
        max_updates_per_run=10,
    ),
    add_learnings_to_context=True,
)

Each store also takes a config object instead of True (UserProfileConfig, EntityMemoryConfig, and so on), and LearningMode controls when writes happen: always, agentic, propose, or hitl. decision_log is the one to reach for first — it is the “why did the agent do that” audit trail people otherwise build by hand.

8.2.4.4 Replay and forking

Older comparisons of Agno — including earlier printings of this book — said it had session storage but no state-snapshot replay, and pointed at Burr for debugging a run that branched wrong. Agno 3.0 closed that gap:

1agent = Agent(db=db, checkpoint="runs")

output = agent.run("Draft the launch post.", session_id="s-1")

2branched = agent.continue_run(
    run_id=output.run_id,
3    continue_from=4,
    fork=True,
    additional_instructions="Try a shorter, punchier opening.",
)

4alt_session = agent.fork_session(source_session_id="s-1")

1

checkpoint sets snapshot granularity — "runs", "tool-batch", or "tools".

2

continue_run resumes, redoes, or branches an earlier run. regenerate=True redoes from the last checkpoint; replace_original controls whether the original is overwritten.

3

continue_from takes a message index, or "end" / "last_user".

4

fork_session deep-copies a whole session into a new one with fresh session_id and fresh run_ids, leaving the source untouched.

RunOutput records the lineage in forked_from_run_id, forked_from_message_index, forked_from_session_id, and regenerated_from. The remaining difference from Burr is not whether you can fork but what you fork into: Burr hands you an explicit typed State and an explicit transition graph, so a forked run is inspectable as data. Agno’s forked run is a message history. For debugging a prompt, the message history is enough; for debugging a state machine, it is not.

Notecontinue_run is undocumented in 3.0.0

The method carries no docstring. The parameter list above is from the signature, which is the contract; edge-case behavior — what replace_original=None resolves to, how fork and regenerate interact — should be checked against agno/agent/_run.py rather than guessed at.

8.2.5 Multi-agent Teams

Agno’s Team primitive handles multi-agent coordination. The key design choice is the mode parameter, which determines how the leader agent coordinates its members:

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.team import Team
from agno.team.mode import TeamMode

researcher = Agent(
    name="Researcher",
1    role="Research specialist who finds and summarizes information",
    model=Claude(id="claude-sonnet-4-6"),
    instructions=[
        "You are a research specialist.",
        "Provide clear, factual summaries on any topic.",
        "Organize findings with structure and cite limitations.",
    ],
)

writer = Agent(
    name="Writer",
2    role="Content writer who crafts polished, engaging text",
    model=Claude(id="claude-sonnet-4-6"),
    instructions=[
        "You are a skilled content writer.",
        "Transform raw information into well-structured, readable text.",
        "Use headers, bullet points, and clear prose.",
    ],
)

team = Team(
3    [researcher, writer],
    name="Research & Writing Team",
4    mode=TeamMode.coordinate,
5    model=Claude(id="claude-sonnet-4-6"),
    instructions=[
        "You lead a research and writing team.",
        "For informational requests, ask the Researcher first,",
        "then ask the Writer to polish the findings.",
        "Synthesize everything into a cohesive response.",
    ],
6    show_members_responses=True,
    markdown=True,
)

team.print_response(
    "Write a brief overview of how large language models are trained.",
    stream=True,
)

1

role describes the agent to the team leader. The leader uses this to decide which member to delegate to. A member with no role is nearly invisible to delegation — the most common reason a team quietly ignores an agent you added.

2

Each member is a full Agent with its own model, tools, and instructions.

3

members is the first positional parameter and it is required. Team breaks the all-keyword pattern that Agent follows. It also accepts a callable returning a member list, for membership resolved at run time.

4

TeamMode.coordinate is the default. Also available: route, broadcast, and tasks.

5

The team leader has its own model. This is the model that makes the delegation decisions.

6

show_members_responses=True includes each member’s individual response in the output — useful for debugging and transparency.

The four modes:

Mode	What the leader does
coordinate	analyzes, picks members, writes each member’s task, synthesizes the answers. The default.
route	picks exactly one member and hands off
broadcast	sends the same task to every member simultaneously
tasks	breaks the work down autonomously against a shared task list

Members can themselves be Teams, so hierarchies nest with no extra machinery. Other parameters worth knowing: respond_directly (member output goes straight to the user with no leader synthesis), determine_input_for_members (default True — the leader rewrites each member’s input), max_iterations=10, and add_team_history_to_members.

8.2.5.1 A worked Team example: research pipeline

Here’s a more realistic team that combines MCP tools, persistence, and memory:

from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.team import Team
from agno.team.mode import TeamMode
from agno.tools.mcp import MCPTools
from agno.db.postgres import PostgresDb

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

web_researcher = Agent(
    name="Web Researcher",
    role="Searches the web for current information",
    model=Claude(id="claude-sonnet-4-6"),
    tools=[MCPTools("npx -y @modelcontextprotocol/server-brave-search",
                    env={"BRAVE_API_KEY": "${BRAVE_API_KEY}"})],
    instructions="Search for current, authoritative sources. Cite URLs.",
)

code_analyst = Agent(
    name="Code Analyst",
    role="Analyzes code repositories and technical documentation",
    model=Claude(id="claude-sonnet-4-6"),
    tools=[MCPTools("npx -y @modelcontextprotocol/server-filesystem /tmp")],
    instructions="Analyze code structure, dependencies, and patterns.",
)

report_writer = Agent(
    name="Report Writer",
    role="Synthesizes research into polished reports",
    model=Claude(id="claude-sonnet-4-6"),
    instructions="Write clear, structured reports with executive summaries.",
)

research_team = Team(
    [web_researcher, code_analyst, report_writer],
    name="Technical Research Team",
    mode=TeamMode.coordinate,
    model=Claude(id="claude-sonnet-4-6"),
1    db=db,
2    enable_agentic_memory=True,
    instructions=[
        "Coordinate research tasks across team members.",
        "Use the Web Researcher for current events and trends.",
        "Use the Code Analyst for technical deep-dives.",
        "Always finish with the Report Writer for synthesis.",
    ],
)

research_team.print_response(
    "Analyze the current state of MCP adoption in production agent systems.",
    stream=True,
3    user_id="analyst-42",
)

1

Team-level persistence. The team stores session history and coordination traces in Postgres.

2

Team-level memory. The team remembers facts about users across sessions — the same memory API as individual agents.

3

user_id scopes memories. Different users get different memory stores.

8.2.6 Skills: SKILL.md folders as capability packs

New in 3.0, and directly compatible with the SKILL.md layout Claude Code uses (Section 4.1.3):

from agno.skills import Skills, LocalSkills

1skills = Skills(loaders=[LocalSkills(path="./skills", validate=True)])
2agent = Agent(model=Claude(id="claude-sonnet-4-6"), skills=skills)

1

A loader points at a directory of skill folders. Each folder holds a SKILL.md with YAML frontmatter (name, description, optional license, compatibility, allowed_tools, arbitrary metadata) plus optional scripts/ and references/.

2

The agent gets three tools — get_skill_instructions, get_skill_reference, get_skill_script — plus a <skills_system> block in its prompt naming what’s available.

The design point is the same as Chapter 4’s: only the skill names and descriptions sit in context, and the body loads on demand when the model asks for it. Skills exposes get_skill_names(), get_skill(name), get_all_skills(), get_tools(), get_system_prompt_snippet(), and reload(). Parse and validation failures raise typed errors — SkillParseError, SkillValidationError.

This is worth pausing on: a skill folder written for Claude Code loads into an Agno agent unmodified. The prompt-pack format is becoming portable across runtimes.

8.2.7 Guardrails and evals as hooks

from agno.guardrails import (
    PIIDetectionGuardrail, PromptInjectionGuardrail, OpenAIModerationGuardrail,
)

agent = Agent(
    model=Claude(id="claude-sonnet-4-6"),
1    pre_hooks=[
        PIIDetectionGuardrail(
            mask_pii=True,
2            custom_patterns={"employee_id": r"EMP-\d{6}"},
        ),
        PromptInjectionGuardrail(),
    ],
    post_hooks=[...],
)

1

pre_hooks and post_hooks accept Callable, BaseGuardrail, or BaseEval — validation, moderation, and scoring share one slot. Subclass BaseGuardrail for custom checks instead of wrapping run().

2

The shipped detectors are configurable, with per-category switches (enable_ssn_check, enable_credit_card_check, enable_email_check, enable_phone_check) and custom regex.

Evals live at agno.eval — accuracy, performance, reliability, and agent_as_judge, all under BaseEval — and results persist to the db’s eval_table. An eval in post_hooks scores production traffic; the same class in a test scores a fixture set.

8.2.8 AgentOS: serving agents as APIs

Agno’s AgentOS wraps agents, teams, and workflows into a deployable application:

from agno.os import AgentOS

agent_os = AgentOS(
    name="research-system",
    description="Production research system",
    agents=[support_agent],
    teams=[research_team],
    workflows=[article_pipeline],
    db=db,
1    tracing=True,
)

2app = agent_os.get_app()

if __name__ == "__main__":
3    agent_os.serve(app="main:app", port=7777, reload=True)

1

tracing=True enables OpenTelemetry tracing across all agents, teams, and workflows.

2

get_app() returns a fastapi.FastAPI. You get REST endpoints for every agent and team automatically.

3

serve() is a uvicorn wrapper. Pass the import string ("main:app"), not the object, when reload=True — same rule as uvicorn itself.

The constructor is where the platform claim gets concrete. Each of these parameters replaces a subsystem teams routinely build:

Parameter	What you no longer write
mcp_server=True	your agents become MCP tools other clients can call
a2a_interface=True	an agent-to-agent protocol endpoint
interfaces=[...]	Slack, Telegram, WhatsApp, AGUI front-ends (agno.os.interfaces.*)
authorization=True, authorization_config, mcp_auth	auth, scopes, and service accounts
scheduler=True, scheduler_poll_interval	cron-style scheduled runs
queue=QueueConfig(...)	a background job queue
event_stream=...	run-event fan-out to subscribers
checkpoint="runs"	OS-wide replay granularity
auto_provision_dbs=True	table creation and migration
base_app=..., on_route_conflict=...	mounting into an existing FastAPI app

TipThe AgentOS-as-MCP-server angle

AgentOS(mcp_server=True) makes Agno the server side of MCP: its agents become tools that other agents — yours, a Claude Code session, or an entirely different framework — can call. This is the inverse of MCPTools, and it is rare among frameworks. The pattern is useful when you’ve built a specialist agent and want it reachable from a stack that has nothing to do with Agno.

mcp_tools = MCPTools(
    transport="streamable-http",
    url="https://my-agent-os.example.com/mcp",
)

8.2.9 OpenTelemetry observability

Observability is built in via the openinference-instrumentation-agno package:

# Install: pip install openinference-instrumentation-agno

from agno.os import AgentOS

agent_os = AgentOS(
    description="Traced system",
    agents=[agent],
1    tracing=True,
)

1

One flag enables tracing. Spans cover agent runs, tool calls, team coordination, knowledge queries, and workflow steps.

The traces follow the OpenInference semantic convention — meaning they work with any OpenTelemetry-compatible backend (Jaeger, Grafana Tempo, Datadog) plus Arize Phoenix for LLM-specific analysis. Spans include:

• Agent run spans — model, input, output, latency, token counts
• Tool call spans — which tool, arguments, return value, duration
• Team coordination spans — leader decisions, member delegations, synthesis
• Knowledge query spans — search terms, filters, retrieved chunks, relevance scores

Traces flow to whichever OpenTelemetry backend you wire up. The session database (SqliteDb, PostgresDb) stores sessions, memories, metrics, and eval results, and it does carry traces_table and spans_table — but querying traces is the backend’s job, not the db API’s. For local development the simplest setup is Arize Phoenix as a local OTel collector:

# pip install arize-phoenix
import phoenix as px

px.launch_app()   # local UI at http://localhost:6006
# The OpenInference instrumentation enabled by `tracing=True` above
# pushes spans to Phoenix automatically; query and visualize from the UI.

For production, point the OTel exporter at Jaeger, Grafana Tempo, Datadog, or Arize’s hosted backend.

8.2.10 Migrating from Agno 1.x and 2.x

Agno moves fast and breaks public API across majors. If you inherit an Agno codebase, this table turns the error message into the fix:

Symptom	Cause	Fix in 3.0
TypeError: run() got an unexpected keyword argument 'message'	1.x run(message=...)	run(input=...), positional
response_model rejected	1.x name	output_schema=
ImportError: cannot import name 'MultiMCPTools'	2.x class, removed	several MCPTools, connect each yourself
ImportError: `mcp` not installed — with mcp installed	mcp 2.x present; Agno needs mcp<2	pip install "agno[mcp]"
agno.storage.* not found	1.x storage module	agno.db.*, and the parameter is db=
AgentKnowledge / *KnowledgeBase not found	1.x knowledge classes	one Knowledge class plus readers
knowledge.load(...) missing	1.x/2.x ingestion	insert() / insert_many()
a Memory(...) object to construct	1.x/2.x memory object	db= plus enable_agentic_memory / update_memory_on_run, or LearningMachine
Team() fails with no arguments	members became required and positional	Team([a, b], ...)
from agno.workflow.workflow import Workflow	2.x deep path	from agno.workflow import Workflow, Step, Condition, Loop, Parallel, Router
from agno.team.team import Team	2.x deep path	from agno.team import Team

New in 3.0 with no 2.x equivalent: skills=, learning=, checkpoint= with continue_run(fork=...) and fork_session(), human_review= on workflow steps, guardrails and evals in pre_hooks/post_hooks, fallback_models, compress_tool_results and offload_tool_results, and the AgentOS scheduler / queue / a2a_interface parameters.

8.2.11 When Agno is the right call

Agno wins when you want multi-agent + MCP + RAG + a control plane on day one and don’t want to spend a sprint assembling them from independent libraries. The cost is opinionation: Knowledge decides how retrieval works, Team decides how routing works, LearningMachine decides what memory looks like. If those defaults match your needs, Agno is the fastest path to a working multi-agent system.

Before writing anything in an Agno codebase, check whether the framework already has it. This list is the most useful thing in the section:

You’re about to build	Agno already ships
a retry/backoff wrapper	retries, delay_between_retries, exponential_backoff
provider failover	fallback_models
a retrieve-then-prompt RAG pipeline	knowledge= plus search_knowledge (already True)
a PDF/website ingestion script	Knowledge readers
a conversation-history trimmer	add_history_to_context + num_history_runs
a “remember the user” table	enable_agentic_memory, update_memory_on_run
an approval queue	@tool(requires_confirmation=True), HumanReview, AgentOS approvals_table
a router picking one specialist	TeamMode.route, or Router in a workflow
a bounded retry loop over a stage	Loop(max_iterations, end_condition)
branch-and-compare debugging	fork_session(), continue_run(fork=True)
a REST API around your agents	AgentOS.get_app()
a Slack or Telegram bot shell	agno.os.interfaces.*
a cron runner	AgentOS(scheduler=True)
PII scrubbing, injection detection	agno.guardrails in pre_hooks
an LLM-as-judge harness	agno.eval.agent_as_judge
OTel spans around runs and tools	tracing=True
a “why did it decide that” log	LearningMachine(decision_log=True)

Agno’s strengths:

• Breadth of integration. MCP tools, 20 vector DBs, 15 storage backends, 141 built-in toolkit modules, 48 model providers — all with consistent APIs.
• Multi-agent out of the box. Team modes handle the common multi-agent patterns without you writing routing logic.
• AgentOS deployment. From prototype to API in minutes, with MCP-server exposure, scheduling, auth, and tracing already wired.
• Agentic RAG. Knowledge-as-a-tool means the model decides when and how to search.
• Human review as a parameter. HumanReview on a step and requires_confirmation on a tool cover the approval cases without a separate service.

Agno’s weaknesses:

• Opinionated abstractions. If Knowledge doesn’t match your retrieval needs, or Team coordination doesn’t fit, you’re fighting the framework rather than using it.
• Less explicit control flow. The Team leader routes via the model — powerful but less predictable than an explicit state machine. For critical paths, use Workflow with executor= callables, or Burr.
• Control flow is Agno’s vocabulary, not yours. A Loop is a loop and a Condition is a branch. Burr routes to any action on any predicate; Agno gives you the shapes that ship.
• State is a message history, not a typed value. Forking works (Section 8.2.4.4), but what you fork is a conversation, not an inspectable State object. When the state machine is the specification, Burr’s typing earns its keep.
• API churn. Three majors in about two years, each with removals. Pin the version and read Section 8.2.10 before upgrading.

8.3 mcp-agent: MCP-Native and Pattern-First

mcp-agent (Last Mile AI) takes a position Agno doesn’t: the Model Context Protocol is the substrate, and an agent framework’s main job is to handle MCP plumbing and ship the canonical patterns from Anthropic’s Building Effective Agents post. If you’ve ever hand-rolled MCP server lifecycle code — process management, OAuth, sampling, elicitation, notifications — you know the work mcp-agent absorbs.

8.3.1 AugmentedLLM and the pattern factories

The core abstraction is AugmentedLLM — a composable wrapper around a model client that holds tools, MCP server connections, and shared context. The patterns ship as factory helpers that return AugmentedLLM objects, which means patterns are AugmentedLLMs and can be nested into larger patterns.

The simplest mcp-agent application:

import asyncio
from mcp_agent.app import MCPApp
from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.llm.augmented_llm_anthropic import AnthropicAugmentedLLM

app = MCPApp(name="simple-agent")

async def main():
1    async with app.run() as running_app:
        agent = Agent(
            name="assistant",
            instruction="You are a helpful assistant.",
2            server_names=["filesystem"],
        )

3        async with agent:
4            llm = await agent.attach_llm(AnthropicAugmentedLLM)
            result = await llm.generate_str("List the files in /tmp")
            print(result)

asyncio.run(main())

1

app.run() provides the runtime context — secrets, OTel exporter, executor, token counter, MCP server registry. All patterns share this context.

2

server_names references MCP servers from the config. The agent gets all tools those servers expose.

3

async with agent manages the agent’s lifecycle — connecting to MCP servers, attaching tools, and cleaning up.

4

attach_llm binds a specific LLM provider to the agent. The AugmentedLLM gets the agent’s tools and instructions.

8.3.2 The pattern library

The pattern library covers Anthropic’s canonical set. Each pattern is a factory function that returns an AugmentedLLM, which means patterns compose — a router can contain a parallel pattern that contains an evaluator-optimizer.

flowchart LR
    A["AugmentedLLM<br/>(base)"] --> R["Router"]
    A --> P["Parallel<br/>(fan-out / fan-in)"]
    A --> E["Evaluator-<br/>Optimizer"]
    A --> O["Orchestrator"]
    A --> S["Swarm"]
    R --> C["Composed<br/>pipelines"]
    P --> C
    E --> C
    style A fill:#f3e8ff,stroke:#6b21a8,color:#6b21a8
    style C fill:#fce7f3,stroke:#9d174d,color:#9d174d

mcp-agent’s pattern library — all patterns are AugmentedLLMs

8.3.2.1 Router pattern

The Router classifies incoming requests and routes to the best-matched agent, function, or MCP server:

from mcp_agent.app import MCPApp
from mcp_agent.agents.agent import Agent, AgentSpec
from mcp_agent.workflows.factory import create_router_llm

app = MCPApp(name="support_router")

async def main():
    async with app.run() as running_app:
1        router = await create_router_llm(
            name="support_router",
2            server_names=["filesystem", "fetch"],
            agents=[
3                AgentSpec(
                    name="finder",
                    instruction="Locate relevant files or URLs.",
                    server_names=["filesystem", "fetch"],
                ),
                AgentSpec(
                    name="writer",
                    instruction="Draft polished responses.",
                ),
            ],
4            functions=[
                lambda msg: "Fallback: escalate to human triage.",
            ],
            routing_instruction="Prefer agents when tool use is needed.",
5            provider="anthropic",
            context=running_app.context,
        )

        # Route and get ranked results
6        decisions = await router.route(
            "Read the contents of README.md",
            top_k=2,
        )
        for choice in decisions:
            print(f"-> {choice.result} "
                  f"(confidence: {choice.confidence}, "
                  f"reasoning: {choice.reasoning})")

        # Or use the router as an autonomous LLM
7        answer = await router.generate_str(
            "Find README.md and summarize it"
        )
        print(answer)

1

create_router_llm returns a Router AugmentedLLM. The router decides where to send each request.

2

server_names are MCP servers the router can route to directly (without wrapping in an agent).

3

AgentSpec describes agents lazily — they’re only instantiated when the router selects them.

4

Functions are simple Python callables — the router can route to them for lightweight tasks.

5

provider selects the LLM backend for routing decisions. Each agent can use a different provider.

6

router.route() returns ranked results with confidence scores and reasoning — useful when you want to inspect the routing decision.

7

router.generate_str() auto-routes and executes in one call — the router picks the best destination and runs it.

8.3.2.2 Parallel (fan-out / fan-in) pattern

Fan out to N agents, fan in with a reducer:

from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.factory import create_parallel_llm

async def main():
    async with app.run() as running_app:
        proofreader = Agent(
            name="proofreader",
            instruction="Review for grammar, spelling, and punctuation.",
        )
        fact_checker = Agent(
            name="fact_checker",
            instruction="Verify factual consistency and logical coherence.",
1            server_names=["fetch"],
        )
        style_enforcer = Agent(
            name="style_enforcer",
            instruction="Analyze narrative flow, tone, and style.",
        )

2        grader = Agent(
            name="grader",
            instruction="""Compile feedback from all reviewers.
            Summarize issues, provide recommendations, assign a grade.""",
        )

3        parallel = create_parallel_llm(
            fan_out=[proofreader, fact_checker, style_enforcer],
            fan_in=grader,
            provider="anthropic",
            context=running_app.context,
        )

4        report = await parallel.generate_str(
            "Grade this story submission:\n\n" + story_text
        )
        print(report)

1

Fan-out agents can have MCP tools. The fact checker can fetch URLs to verify claims.

2

The fan-in agent synthesizes all fan-out results into a single output.

3

create_parallel_llm returns a Parallel AugmentedLLM. Fan-out agents run concurrently via asyncio.gather.

4

One call, three parallel agents, one synthesis. The generate_str() interface is the same as any other AugmentedLLM.

8.3.2.3 Evaluator-Optimizer pattern

Generate, critique, refine — in a loop:

from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.factory import create_evaluator_optimizer_llm

async def main():
    async with app.run() as running_app:
        writer = Agent(
            name="writer",
            instruction="Write compelling marketing copy. "
                        "Incorporate feedback from previous rounds.",
        )

        critic = Agent(
            name="critic",
            instruction="""Evaluate marketing copy on clarity, engagement,
            and persuasiveness. Rate 1-10 and provide specific
            improvement suggestions.""",
        )

        eval_opt = create_evaluator_optimizer_llm(
1            optimizer=writer,
2            evaluator=critic,
3            min_rating=8,
4            max_refinements=3,
            provider="anthropic",
            context=running_app.context,
        )

        result = await eval_opt.generate_str(
            "Write a tagline for a new AI coding assistant"
        )
        print(f"Final copy (after refinement): {result}")

1

Optimizer generates candidates. On each iteration, it sees the critic’s feedback and tries to improve.

2

Evaluator critiques. Returns a rating and specific improvement suggestions.

3

min_rating=8 is the quality threshold. The loop stops when the evaluator rates the output at 8 or above.

4

max_refinements=3 caps the iterations. If the optimizer can’t reach the threshold in 3 rounds, return the best attempt.

NoteThe Evaluator-Optimizer is the Ralph loop in disguise

The Evaluator-Optimizer pattern is structurally the same as the Ralph loop you’ll meet in Chapter 11: generate a candidate, evaluate against criteria, iterate. Ralph applies this shape to code editing — produce a change, run tests, iterate. The pattern is the same; the action space is different.

8.3.2.4 Composing patterns

Because every pattern is an AugmentedLLM, you compose them like building blocks:

from mcp_agent.workflows.factory import (
    create_router_llm,
    create_parallel_llm,
    create_evaluator_optimizer_llm,
)

async def main():
    async with app.run() as running_app:
        # Step 1: Route to the right research strategy
        router = await create_router_llm(
            agents=[web_researcher, code_analyst, doc_reader],
            provider="anthropic",
            context=running_app.context,
        )

        # Step 2: Fan out for multi-perspective analysis
        parallel = create_parallel_llm(
            fan_out=[technical_reviewer, business_reviewer],
            fan_in=synthesizer,
            provider="anthropic",
            context=running_app.context,
        )

        # Step 3: Iteratively refine the output
        eval_opt = create_evaluator_optimizer_llm(
            optimizer=writer,
            evaluator=editor,
            min_rating=8,
            max_refinements=3,
            provider="anthropic",
            context=running_app.context,
        )

        # Compose: route -> parallel -> evaluate
1        research = await router.generate_str(query)
2        analysis = await parallel.generate_str(research)
3        final = await eval_opt.generate_str(analysis)

1

Router selects the research strategy based on the query.

2

Parallel fans out to multiple reviewers and synthesizes their analysis.

3

Evaluator-Optimizer refines the synthesis until the editor is satisfied.

8.3.3 What MCP-native actually buys you

The “MCP-native” claim is concrete. mcp-agent implements the MCP spec more completely than any other agent framework:

• Lifecycle management — server processes are started, supervised, and torn down by the framework. You write a YAML config or Settings object; mcp-agent runs the servers.
• Auth — OAuth flows for MCP servers that require it. The framework handles token refresh and credential storage.
• Elicitation and sampling — the bidirectional MCP capabilities where the server asks the client for input mid-tool-call. mcp-agent implements both sides.
• Notifications — server-pushed events make it through the runtime.

Server configuration is declarative — YAML or Python:

# mcp_agent.config.yaml
mcp:
  servers:
    brave-search:
      command: "npx"
      args: ["-y", "@modelcontextprotocol/server-brave-search"]
      env:
        BRAVE_API_KEY: "${BRAVE_API_KEY}"
    filesystem:
      command: "npx"
      args: ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]

anthropic:
  default_model: "claude-sonnet-4-6"

Or equivalently in Python:

from mcp_agent.config import Settings, MCPSettings, MCPServerSettings

settings = Settings(
1    execution_engine="asyncio",
    mcp=MCPSettings(
        servers={
            "brave-search": MCPServerSettings(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-brave-search"],
                env={"BRAVE_API_KEY": "${BRAVE_API_KEY}"},
            ),
            "filesystem": MCPServerSettings(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
            ),
        }
    ),
)

1

execution_engine can be "asyncio" (default) or "temporal" for durable execution — see Section 8.3.4.

The other frameworks treat MCP as one transport among many. Agno’s MCPTools is capable but MCP is one tool type among many native tools. For mcp-agent, “MCP is all you need” is the founding claim — tools come from MCP servers, and the framework’s job is to make that plumbing invisible.

8.3.4 Durable execution via Temporal

mcp-agent integrates with Temporal for durable execution. Switch one config line and your agent workflows become crash-recoverable:

from mcp_agent.app import MCPApp
from mcp_agent.executor.workflow import Workflow, WorkflowResult
from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.llm.augmented_llm_anthropic import AnthropicAugmentedLLM
import asyncio

app = MCPApp(name="durable_analysis")

1@app.workflow
class ParallelAnalysisWorkflow(Workflow[dict]):
2    @app.workflow_run
    async def run(self, document: str) -> WorkflowResult[dict]:
        async def analyze_sentiment():
            agent = Agent(name="sentiment",
                          instruction="Analyze sentiment.")
            async with agent:
                llm = await agent.attach_llm(AnthropicAugmentedLLM)
                return await llm.generate_str(
                    f"Analyze sentiment: {document}")

        async def extract_entities():
            agent = Agent(name="entities",
                          instruction="Extract entities.")
            async with agent:
                llm = await agent.attach_llm(AnthropicAugmentedLLM)
                return await llm.generate_str(
                    f"Extract entities: {document}")

        async def summarize():
            agent = Agent(name="summarizer",
                          instruction="Summarize content.")
            async with agent:
                llm = await agent.attach_llm(AnthropicAugmentedLLM)
                return await llm.generate_str(
                    f"Summarize: {document}")

        # Execute in parallel — Temporal handles orchestration
3        sentiment, entities, summary = await asyncio.gather(
            analyze_sentiment(),
            extract_entities(),
            summarize()
        )

4        return WorkflowResult(value={
            "sentiment": sentiment,
            "entities": entities,
            "summary": summary
        })

1

@app.workflow registers a workflow class. With execution_engine="temporal", this becomes a Temporal workflow.

2

@app.workflow_run marks the entry point. Temporal records the event history at each await.

3

asyncio.gather fans out. With Temporal, each branch is a separate activity — if one fails, only that branch retries.

4

WorkflowResult wraps the output. Temporal persists this — the result survives worker restarts.

The mental model contrast with Burr’s fork-from-state (Chapter 7):

	Burr (fork-from-state)	mcp-agent (Temporal replay)
What’s saved	State snapshot after each action	Event history of the workflow
Primary use	Debug-and-rerun-with-changes	Crash recovery + exactly-once execution
Granularity	Per-action snapshot	Per-event in the workflow
“What if?” branching	First-class via fork_from_sequence_id	Possible but not the primary mode

Both are real production guarantees; they solve adjacent but different problems. If your need is “this agent must complete even if the worker pod is killed,” Temporal. If your need is “this agent went wrong and I want to know which decision and try a different one,” Burr.

8.3.5 When mcp-agent is the right call

mcp-agent’s strengths:

• Full MCP spec implementation. If you have nontrivial MCP server plumbing — OAuth, elicitation, sampling, server lifecycle — mcp-agent handles it. No other framework is as complete here.
• Pattern composition. The factory-function-returning-AugmentedLLM model means patterns nest. A router-inside-a-parallel-inside-an-evaluator is three factory calls, not three frameworks.
• Temporal for durability. Crash-recoverable agent workflows with exactly-once semantics. Essential for long-running agents in production.
• Lightweight agents. Agents are thin — name, instruction, server names. The complexity lives in the patterns, not the agent configuration.

mcp-agent’s weaknesses:

• No Knowledge primitive. RAG is DIY — you bring an MCP server that provides retrieval, or you wire your own vector search. Agno’s Knowledge is far more convenient.
• No built-in memory. Session history, user preferences, persistent facts — all yours to implement. Agno’s memory story is significantly richer.
• No deployment story. mcp-agent is a library, not a platform. You write your own FastAPI wrapper, your own Docker deployment, your own RBAC. Agno’s AgentOS ships these.
• Less explicit state. The pattern composition is implicit — there’s no state machine you can inspect or replay. Burr is the tool for that.

8.4 Worked Example: Q&A Agent, Two Ways

Chapter 7 built a Q&A agent in PocketFlow: a DecideAction node returning "search" or "answer", a SearchWeb node, and an Answer node. We’ll port it to both frameworks. The comparison shows how each framework’s center of gravity shapes the same problem differently.

8.4.1 Agno version

The Agno version leans on Knowledge for retrieval and Team for multi-step coordination:

import asyncio
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools.mcp import MCPTools
from agno.team import Team
from agno.team.mode import TeamMode
from agno.db.sqlite import SqliteDb

db = SqliteDb(db_file="qa_agent.db")

async def main():
    # MCP-based web search
    search_tools = MCPTools(
        "npx -y @modelcontextprotocol/server-brave-search",
        env={"BRAVE_API_KEY": "${BRAVE_API_KEY}"},
    )
    await search_tools.connect()

1    researcher = Agent(
        name="Researcher",
        role="Search the web for information to answer questions",
        model=Claude(id="claude-sonnet-4-6"),
        tools=[search_tools],
        instructions=[
            "Search for authoritative sources to answer the question.",
            "Return a structured brief with key findings and source URLs.",
        ],
    )

2    synthesizer = Agent(
        name="Synthesizer",
        role="Synthesize research into a clear, grounded answer",
        model=Claude(id="claude-sonnet-4-6"),
        instructions=[
            "Answer the user's question using only the research provided.",
            "Cite sources. If the evidence is insufficient, say so.",
        ],
    )

3    qa_team = Team(
        [researcher, synthesizer],
        name="QA Team",
        mode=TeamMode.coordinate,
        model=Claude(id="claude-sonnet-4-6"),
        db=db,
        instructions=[
            "For factual questions, delegate to the Researcher first.",
            "Then ask the Synthesizer to produce the final answer.",
            "If the question is conversational, answer directly.",
        ],
        show_members_responses=True,
        markdown=True,
    )

    qa_team.print_response(
        "Who won the 2024 Physics Nobel and what was it for?",
        stream=True,
    )
    await search_tools.close()

asyncio.run(main())

1

Researcher agent has web search via MCP. It gathers evidence.

2

Synthesizer agent takes the researcher’s findings and produces a grounded answer.

3

Team in coordinate mode. Members go in positionally. The leader decides the pipeline: research first, synthesize second. The routing is model-driven, not hard-coded.

The PocketFlow version had explicit DecideAction -> SearchWeb -> Answer nodes with action edges. The Agno version replaces the explicit state machine with a team leader that makes the same decisions — but the routing logic is in the model, not the graph.

8.4.2 mcp-agent version

The mcp-agent version uses the Router pattern with an MCP search server. Note that MCPApp(settings=...) accepts a Settings object or a YAML path — not a raw dict; passing {...} here would fail at attribute access time:

import asyncio
from mcp_agent.app import MCPApp
from mcp_agent.agents.agent import Agent
from mcp_agent.workflows.factory import create_router_llm
from mcp_agent.config import Settings, MCPSettings, MCPServerSettings

app = MCPApp(
    name="qa-bot",
    settings=Settings(
        mcp=MCPSettings(
            servers={
1                "brave-search": MCPServerSettings(
                    command="npx",
                    args=["-y", "@modelcontextprotocol/server-brave-search"],
                    env={"BRAVE_API_KEY": "${BRAVE_API_KEY}"},
                ),
            }
        )
    ),
)

async def main():
    async with app.run() as running_app:
2        researcher = Agent(
            name="researcher",
            instruction="Use search to gather evidence. "
                        "Return a structured brief with sources.",
            server_names=["brave-search"],
        )

3        synthesizer = Agent(
            name="synthesizer",
            instruction="Answer the user's question grounded "
                        "in the provided research.",
        )

4        direct_answerer = Agent(
            name="direct_answerer",
            instruction="Answer simple conversational questions "
                        "that don't require research.",
        )

5        router = await create_router_llm(
            agents=[researcher, synthesizer, direct_answerer],
            routing_instruction="Route factual questions to researcher, "
                               "then synthesizer. Route conversational "
                               "questions to direct_answerer.",
            provider="anthropic",
            context=running_app.context,
        )

6        answer = await router.generate_str(
            "Who won the 2024 Physics Nobel and what was it for?"
        )
        print(answer)

asyncio.run(main())

1

MCP server is declarative. The Brave Search server is started, supervised, and authenticated by mcp-agent.

2

Researcher agent gets the MCP server. It has tools; the others don’t.

3

Synthesizer receives the researcher’s output and produces the grounded answer.

4

Direct answerer handles simple queries without research — the router skips the expensive path.

5

create_router_llm returns a Router AugmentedLLM. The routing decision happens inside generate_str.

6

One call — the router picks the agent, runs it, and returns the result.

The shape difference is the actual decision the comparison is helping you make:

• Agno’s Team makes routing decisions via a leader agent that sees the full team and coordinates explicitly. You get show_members_responses=True for transparency and session persistence for free.
• mcp-agent’s Router makes routing decisions via classification and confidence scoring. You get composability — this router can be dropped into a larger pattern (a parallel-then-router-then-evaluator stack) without rewriting.

8.5 The Decision Matrix

Axis	Agno	mcp-agent
Center of gravity	Batteries-included agent platform	MCP-spec-native + Anthropic patterns
Killer feature	Knowledge + Team + MCPTools + AgentOS control plane	Full MCP spec + pattern factories + Temporal
Primitives	Agent, Team, Workflow, Knowledge	Agent, AugmentedLLM, pattern factories
MCP story	First-class MCPTools + AgentOS-as-server	The substrate — lifecycle, auth, elicitation, sampling
RAG	Knowledge primitive, 20 vector DBs	DIY (via MCP servers)
Multi-agent	Native Team with 4 coordination modes	Nested AugmentedLLMs + Swarm pattern
Memory	Agentic memory + LearningMachine + 15 DB backends	DIY
Replay / durability	checkpoint= + continue_run(fork=…) + fork_session() (3.0)	Temporal workflow replay
Observability	OTel via OpenInference, built-in trace DB	OTLP + TokenCounter
Human review	HumanReview per step, requires_confirmation per tool	Elicitation via MCP
Deployment	AgentOS (FastAPI, RBAC, MCP server, scheduler, queue)	Library only (bring your own deployment)
Governance	Commercial (Agno) + OSS	Last Mile AI + OSS
Pick when…	You want MCP + RAG + Team + memory without assembly	You’re MCP-first, want patterns as primitives, need durable execution

For the full three-way comparison (including Burr + Pydantic), see the Burr sections in Chapter 7. Agno 3.0 closed the replay gap that earlier comparisons put in Burr’s column (Section 8.2.4.4); what still separates them is typed state — Burr forks an inspectable State, Agno forks a message history.

TipCan you combine them?

Agno + mcp-agent overlap significantly — both want to own the agent surface — so picking one is usually cleaner.

Burr + Agno is plausible: use Burr’s state machine for the conversational loop (Chapter 7), call an Agno agent as a tool inside an action. The composition isn’t documented but the interfaces are loose enough.

Burr + mcp-agent is similar: a Burr action can wrap an mcp-agent AugmentedLLM and benefit from both typed-state replay and the pattern library. This is the most synergistic pairing — Burr for the outer loop and state inspection, mcp-agent for the inner pattern composition.

NoteWhat about Mastra?

Chapter 9 covers Mastra — a TypeScript-native agent framework. If your team is TypeScript-first, Mastra is the production framework to evaluate. The Python-vs-TypeScript choice often dominates the framework choice within each language.

8.6 Forward Link: From Agent Platforms to the Ralph Loop

Chapter 11 introduces Ralph — Geoffrey Huntley’s minimal autonomous coding loop: load spec, select task, execute, observe, repeat. Ralph and the frameworks in this chapter are different layers of the same stack. Ralph is the loop shape; the frameworks are the state machine the loop runs on.

Three connections worth flagging now:

• mcp-agent’s Evaluator-Optimizer pattern is the Ralph loop in disguise. Generate a candidate change; evaluate against the spec/tests; iterate. Chapter 11 will show that the “spec -> execute -> test -> re-evaluate” loop is structurally the same pattern, just with code edits as the action space.
• Agno’s Team is what scales Ralph to Gas Town. Chapter 13 covers multi-agent orchestration; if you adopted Agno here, the Team primitive is what you’ll lean on then.
• Burr’s @action.pydantic (Chapter 7) is what makes a Ralph harness inspectable. When the loop runs for 200 iterations and one of them edits the wrong file, you want to know which action wrote which field. Typed reads/writes are not optional at that scale.

The book’s path is therefore: Chapter 7 picks the state machine, this chapter picks the platform, Chapter 10 picks the posture, Chapter 11 wraps the loop around it, and Chapter 13 scales the loop into a fleet.

NoteStill can’t choose? That’s the next chapter’s job

Feature tables — including the one in Section 8.5 — stall when the requirement is “this thing takes irreversible actions and I need to explain afterwards what it did.” Section 10.4 reframes the choice around when control flow is decided rather than which features ship, scores all five candidates (adding Pydantic AI and Microsoft Agent Framework) on five axes, and names the one capability none of them provides.

8.7 Key Takeaways

• PocketFlow stops being enough when MCP servers proliferate and you need multi-agent coordination, RAG, and memory. Chapter 7 (Burr + Pydantic) addresses typed state and replay; this chapter addresses the platform question.
• Agno is the batteries-included pick. Agent / Team / Workflow / Knowledge plus MCPTools, multi-backend memory, per-step HumanReview, and the AgentOS control plane ship MCP, RAG, multi-agent, approvals, and deployment as one decision instead of five integrations. The cost is opinionation — Agno decides how retrieval, routing, and memory work — and API churn: three majors in two years, each with removals (Section 8.2.10).
• mcp-agent is the MCP-native pick. The Model Context Protocol is the substrate, AugmentedLLM is the composition primitive, and the canonical Anthropic patterns (Router, Parallel, Evaluator-Optimizer) ship as factory helpers. Temporal handles durable execution. The cost is assembly — no built-in RAG, memory, or deployment story.
• The decision matrix in Section 8.5 is honest about gaps — neither framework is a superset of the other. The right choice depends on which pressure dominates your project.
• Frameworks compose better than they replace. Burr (Chapter 7) + mcp-agent is a real option: typed-state replay for the outer loop, AugmentedLLM patterns inside the actions.
• Pin the Agno version. Every signature in this chapter is agno 3.0.0; 1.x and 2.x code fails at import or at the first keyword argument.

8.8 Concept Map

flowchart TD
    P["Pressures past PocketFlow"] --> M["MCP plumbing"]
    P --> K["Multi-agent + RAG + memory"]
    P --> PT["Canonical agent patterns"]
    M --> AG["Agno (MCPTools)"]
    K --> AG
    M --> MA["mcp-agent (MCP substrate)"]
    PT --> MA
    AG --> TM["Team coordination"]
    AG --> KN["Knowledge (agentic RAG)"]
    AG --> MM["Memory + sessions"]
    MA --> RT["Router pattern"]
    MA --> PL["Parallel pattern"]
    MA --> EO["Evaluator-Optimizer"]
    MA --> TP["Temporal durability"]
    EO --> RL["Ralph loop (Ch 11)"]
    TM --> GT["Gas Town (Ch 13)"]
    style P fill:#fef3c7,stroke:#92400e,color:#92400e
    style AG fill:#dcfce7,stroke:#166534,color:#166534
    style MA fill:#f3e8ff,stroke:#6b21a8,color:#6b21a8
    style RL fill:#fce7f3,stroke:#9d174d,color:#9d174d
    style GT fill:#fce7f3,stroke:#9d174d,color:#9d174d

How the production-platform concepts connect