AETIUS

Agno adapter

Mapping

Agno has three separate native primitives that fit the three classified graph shapes exactly, confirmed against agno's real source before writing this adapter:

  • single_agent maps to a plain agno.agent.Agent. system_message is omitted entirely, not set to the disclosure placeholder, when system_prompt_source == "unavailable". This mirrors a rule already established in this codebase for the same reason given there: passing that placeholder text as the model's literal system prompt would read to the model as real instructions, worse than having none.
  • pipeline maps to agno.workflow.Workflow(steps=[Step(agent=...), ...]), in edge order. Agno's Team modes (coordinate, route, broadcast, tasks) are all leader-delegates-to-members shapes. None of them is "run these roles in a fixed sequence," which is exactly what a pipeline is and exactly what Workflow's Step sequence is for.
  • supervisor_delegates maps to agno.team.Team(mode=TeamMode.coordinate). The hub role becomes the team leader; each delegate becomes a member Agent. coordinate, specifically, not route or broadcast, is the mode whose real behavior, the leader picks members and crafts tasks, matches what edges into more than one distinct delegate actually show: the hub choosing among delegates, not fanning out to all of them at once or handing off permanently to one.
  • unclassified refuses. Agno has no general-graph primitive to fall back to the way LangGraph does; Team is leader-plus-members and Workflow is a fixed step sequence, neither of which can represent an arbitrary graph shape without guessing which one it should be.

Every one of these still needs the developer's own real model. agentstage never constructs one or makes a model call itself. model is a required parameter for exactly that reason.

End to end example

None of the model calls below hit a real API. ScriptedModel is a minimal, real agno.models.base.Model subclass that returns a fixed sequence of responses instead of calling a provider, so these examples run for real, with real tool dispatch through the replay matcher, at zero cost and with no credentials. A real system would pass its own real model in model (and, per role, in member_models) instead.

Python
import json
from agno.models.base import Model
from agno.models.response import ModelResponse


class ScriptedModel(Model):
    def __init__(self, steps, **kw):
        kw.setdefault("id", "scripted")
        kw.setdefault("name", "Scripted")
        kw.setdefault("provider", "scripted")
        super().__init__(**kw)
        self.steps = steps
        self.i = 0

    def invoke(self, *a, **k):
        step = self.steps[min(self.i, len(self.steps) - 1)]
        self.i += 1
        return step

    async def ainvoke(self, *a, **k):
        return self.invoke(*a, **k)

    def invoke_stream(self, *a, **k):
        yield self.invoke(*a, **k)

    async def ainvoke_stream(self, *a, **k):
        yield self.invoke(*a, **k)

    def _parse_provider_response(self, response, **k):
        return response

    def _parse_provider_response_delta(self, response):
        return response


def scripted_tool_call(tool_name, args, final_text):
    return ScriptedModel(
        [
            ModelResponse(
                role="assistant",
                tool_calls=[{"id": "call_1", "type": "function", "function": {"name": tool_name, "arguments": json.dumps(args)}}],
            ),
            ModelResponse(role="assistant", content=final_text),
        ]
    )

single_agent

Python
from examples.fixtures import build_environment
from agentstage.agno import adapt
from agno.agent import Agent

env = build_environment()
invoice_system = env.reconstruct("invoice-agent")
agent = adapt(invoice_system, model=scripted_tool_call("lookup_invoice", {"invoice_id": "inv_1003"}, "Invoice inv_1003 is paid, amount 89.99."))

print(type(agent), isinstance(agent, Agent))
print(agent.system_message)
result = agent.run("What is the status of invoice inv_1003?")
print(result.content)
for t in result.tools or []:
    print(" tool call:", t.tool_name, t.tool_args, "->", t.result)
Output
<class 'agno.agent.agent.Agent'> True
You are an invoicing support assistant. Look up invoices and answer billing questions accurately using the tools provided. Never guess at an invoice's status.
Invoice inv_1003 is paid, amount 89.99.
 tool call: lookup_invoice {'invoice_id': 'inv_1003'} -> {'status': 'paid', 'amount': 89.99}

The scripted model decided to call lookup_invoice; the tool that ran was the real, replay-backed reconstructed one, and its result is a real historical response.

Note on reading t.result: it looks like a dict when printed above, but it is not one.

Python
import ast

execution = result.tools[0]
print(type(execution.result))
print(execution.result)

structured = ast.literal_eval(execution.result)
print(type(structured))
print(structured)
Output
<class 'str'>
{'status': 'paid', 'amount': 89.99}
<class 'dict'>
{'status': 'paid', 'amount': 89.99}

execution.result is a str, specifically Python's str(dict) form of the real replayed value, not the structured dict itself. json.loads does not parse it (it is not valid JSON, the quotes are single, not double); ast.literal_eval does, since the string is valid Python literal syntax. This is a property of how Agno's own ToolExecution stores results, not something agentstage does to it. The value itself is still the real, replay-backed historical response either way.

pipeline

Python
pipeline_system = env.reconstruct("support-pipeline")
from agno.workflow import Workflow

member_models = {
    "intake": scripted_tool_call("classify_ticket", {"text": "my invoice total looks wrong"}, "Classified as billing."),
    "triage": scripted_tool_call("lookup_customer", {"id": "cust_201"}, "Customer is Ada Lovelace, pro plan."),
    "resolver": scripted_tool_call("issue_refund", {"invoice_id": "inv_2001"}, "Refund issued."),
}
workflow = adapt(pipeline_system, model=member_models["intake"], member_models=member_models)
print(type(workflow), isinstance(workflow, Workflow))
print([step.name for step in workflow.steps])
result = workflow.run("my invoice total looks wrong")
print(result.content)
Output
<class 'agno.workflow.workflow.Workflow'> True
['intake', 'triage', 'resolver']
Refund issued.

member_models overrides the model per role, since a reconstructed pipeline may have run different real models at each step. All three steps ran in the observed order, each dispatching its own real tool call.

supervisor_delegates

Python
team_system = env.reconstruct("support-team")
from agno.team import Team
from agno.team.mode import TeamMode

hub_model = ScriptedModel(
    [
        ModelResponse(
            role="assistant",
            tool_calls=[
                {
                    "id": "call_1",
                    "type": "function",
                    "function": {
                        "name": "delegate_task_to_member",
                        "arguments": json.dumps({"member_id": "researcher", "task": "find the refund policy for annual plans"}),
                    },
                }
            ],
        ),
        ModelResponse(role="assistant", content="Refunds on annual plans are available within 30 days per our policy doc."),
    ]
)
researcher_model = scripted_tool_call("search_docs", {"query": "refund policy for annual plans"}, "Found refund-policy.md#annual-plans.")
writer_model = ScriptedModel([ModelResponse(role="assistant", content="unused")])

team = adapt(team_system, model=hub_model, member_models={"researcher": researcher_model, "writer": writer_model})
print(type(team), isinstance(team, Team))
print(team.mode == TeamMode.coordinate)
print(team.id, [m.id for m in team.members])
result = team.run("What is our refund policy for annual plans?")
print(result.content)
Output
<class 'agno.team.team.Team'> True
True
supervisor ['researcher', 'writer']
Refunds on annual plans are available within 30 days per our policy doc.

The hub's scripted decision to delegate to researcher used Agno's own delegate_task_to_member mechanism, exactly what a real leader model would call. The researcher's own scripted model then made a real, replay-backed search_docs call before returning its finding back up to the leader, which synthesized the final answer. writer_model was never used in this run, since the leader only delegated to researcher, the same way a real coordinate-mode team only calls the members it decides it needs.

A Braintrust-sourced system

Everything above used the Langfuse-shaped fixture. agentstage.agno.adapt() does not know or care which source built the ReconstructedSystem it is given; the same ScriptedModel trick works unchanged. No live Braintrust project was available while writing this page either, so this uses examples/braintrust_fixtures.py, the same disclosed-fixture pattern as docs/sources/braintrust.md.

Python
from examples.braintrust_fixtures import build_environment as build_braintrust_environment

bt_env = build_braintrust_environment()
bt_invoice_system = bt_env.reconstruct("invoice-agent")
bt_agent = adapt(bt_invoice_system, model=scripted_tool_call("lookup_invoice", {"invoice_id": "inv_3001"}, "Invoice inv_3001 is paid, amount 145.0."))
print(type(bt_agent), isinstance(bt_agent, Agent))
print(bt_agent.id)
print(bt_agent.system_message)
bt_result = bt_agent.run("What is the status of invoice inv_3001?")
print(bt_result.content)
for t in bt_result.tools or []:
    print(" tool call:", t.tool_name, t.tool_args, "->", t.result)
Output
<class 'agno.agent.agent.Agent'> True
invoice-agent
You are an invoicing support assistant for a Braintrust-logged deployment. Look up invoices accurately.
Invoice inv_3001 is paid, amount 145.0.
 tool call: lookup_invoice {'invoice_id': 'inv_3001'} -> {'status': 'paid', 'amount': 145.0}

One real difference from the Langfuse-sourced example above: bt_agent.id is invoice-agent, not the generic agent fallback the Langfuse-sourced single_agent example used, for the same reason noted on the LangGraph adapter page: Braintrust has no separate trace-level object, so the grouping key and the root span's own resolved role identity both come from the same metadata.agent_name. t.result is still Agno's own stringified form here too, exactly as the note above describes; ast.literal_eval recovers it the same way.