aer1-autogen

AER-1 verifiable swarm receipts for Microsoft AutoGen.

Wrap your agents and the whole swarm run emits one hash-chained, offline-verifiable verifiable workflow receipt (AER-1, Section 8): every agent's turn, every tool call, every handoff, in order, each step carrying the agent's identity, with a Merkle root over the entire run. No network calls, no behavior changes, the wrapper only observes.

Works with autogen-agentchat (AssistantAgent, UserProxyAgent, RoundRobinGroupChat, SelectorGroupChat, Swarm, ...) and with legacy pyautogen 0.x ConversableAgents via observation hooks.

The AER-1 framework collector family

The AutoGen collector in the AER-1 framework collector family. Any agent running on these frameworks can emit verifiable AER-1 execution receipts: every step recorded, hash-chained, one Merkle root over the whole run.

• aer1-smolagents ��� SmolAgents agents via step callbacks
• aer1-openai-agents ��� the OpenAI Agents SDK via its TracingProcessor
• aer1-crewai ��� CrewAI crews via the event bus
• aer1-langgraph ��� LangGraph swarms via callback handler
• aer1-autogen ��� AutoGen multi-agent chats

Plus the aer1 metapackage: pip install aer1, then import aer1; aer1.instrument(). One line, zero config, auto-detects your framework.

Install

pip install aer1-autogen

Use it (three lines)

import asyncio
from aer1_autogen import AER1AgentWrapper, AER1AutoGenCollector
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat

collector = AER1AutoGenCollector(goal="Plan the product launch")   # line 1

async def main():
    alice = AER1AgentWrapper(AssistantAgent("alice", model_client=client), collector)  # line 2
    bob = AER1AgentWrapper(AssistantAgent("bob", model_client=client), collector)
    team = RoundRobinGroupChat([alice, bob], max_turns=6)
    result = await team.run(task="Plan the product launch")
    receipt = collector.finalize(final_answer=result.messages[-1].content)             # line 3
    assert collector.verify(receipt) == []                                             # VALID
    collector.save("swarm-receipt.json", workflow=receipt)

asyncio.run(main())

That is the whole integration. The receipt is a plain JSON object you can store, ship to an auditor, or render in a UI.

What the receipt contains

Workflow level (AER-1 Section 8, Table 2):

• type, version, workflow_id, receipt_id, session_id
• goal, status
• steps: one record per observed event, seq 1..n in order
• merkle_root: Section 8.1 root over the ordered step receipt ids
• output_hash: SHA-256 of the final answer
• verify_url: where the verification procedure is documented

Step level (AER-1 Section 8, Table 3), plus swarm identity extras:

• seq, receipt_id, tool, receipt_hash, started_at, ended_at, status
• agent: the name of the agent that acted (every step)
• reply_to: which agent spoke just before this turn (message steps)
• target: handoff destination (handoff steps)
• recipient: message recipient (legacy pyautogen steps)

Step kinds:

• message: an agent produced a reply. The receipt_hash commits to the message content.
• <tool name>: an agent executed a tool. The hash commits to the tool name, arguments, result, and error flag.
• handoff: an agent handed the conversation to another agent (Swarm teams). The hash commits to the handoff content and target.

Each step receipt_hash is SHA-256 over the canonical JSON of what actually happened. The hash commits to the content; the receipt stays compact.

Legacy pyautogen (0.x)

from aer1_autogen import AER1AutoGenCollector, attach_legacy_hooks
import autogen

collector = AER1AutoGenCollector(goal="Two-agent task")
alice = autogen.ConversableAgent("alice", llm_config=False, human_input_mode="NEVER")
bob = autogen.ConversableAgent("bob", llm_config=False, human_input_mode="NEVER")
attach_legacy_hooks(alice, collector)
attach_legacy_hooks(bob, collector)

alice.initiate_chat(bob, message="Start the task", max_turns=4)
receipt = collector.finalize()
assert collector.verify(receipt) == []  # VALID

This uses pyautogen's own process_message_before_send hook point plus a recording wrapper around execute_function. Needs pip install pyautogen (it is optional; autogen-agentchat is the default dependency).

Verification

collector.verify(receipt) runs the full offline check and returns a list of failure reasons, empty when valid:

• all Table 2 / Table 3 members present and well-formed
• seq values exactly 1..n in order, no gaps
• no two steps share a receipt_id (MM-1)
• merkle_root matches the recomputed Section 8.1 root
• strict RFC 3339 timestamps, lowercase UUIDs, 64-char hex digests

Tamper with any field and verification fails. Try it:

receipt["steps"][0]["tool"] = ""
assert collector.verify(receipt) != []  # fails, as it should

Notes

• The wrapper keeps the inner agent's name and description, so team routing, speaker selection, and handoffs behave exactly as unwrapped.
• on_messages_stream delegates through on_messages, so streaming runs are recorded exactly once.
• session_id defaults to a fresh UUID per collector; pass your own to correlate receipts across runs.
• verify_url defaults to the AER-1 specification page; point it at your own verifier in production.
• The collector never touches the network and never changes what the agents do. It only observes. A recording failure can never break a run.

Spec

AER-1: Agent Execution Receipts, draft-zambo-aer1, https://datatracker.ietf.org/doc/draft-zambo-aer1/

See it live

Your receipt is offline-verifiable, but you can also check it on the live verifier:

1. Copy the receipt JSON your code produced
2. Paste it at https://zambo.dev/verify
3. See the verification result with the Merkle root and step hashes

Or mint a live receipt directly: run any call at https://zambo.dev/demo and get a shareable receipt URL like https://zambo.dev/run/.

License

Apache-2.0