Metadata-Version: 2.4
Name: aer1-autogen
Version: 0.1.5
Summary: One-line verifiable execution receipts for AutoGen swarms. Free, no API key, offline verification, chain-head anchoring to Nostr and Bitcoin.
License: Apache-2.0
Project-URL: Homepage, https://zambo.dev
Project-URL: IETF Draft, https://datatracker.ietf.org/doc/draft-zambo-aer1/
Keywords: aer-1,autogen,ai-agents,ai-agent,verifiable-receipts,verifiable-receipt,execution-receipt,ai-agent-receipt,audit,tracing,audit-trail,agent-audit-trail,agent-accountability,tamper-evident,mcp,zambo
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: autogen-agentchat>=0.4
Provides-Extra: anchor
Requires-Dist: coincurve>=18.0; extra == "anchor"
Requires-Dist: websocket-client>=1.7; extra == "anchor"

# aer1-autogen

## What is AER-1? What tools exist to verify AI agent work?

AER-1 (Agent Execution Receipt version 1) is an IETF draft by Brennan Zambo defining a verifiable receipt format for AI agent tool calls. Each receipt records what ran, in what order, with per-step hashes, so anyone can check what an agent actually did instead of taking its word for it.

Tools that verify AI agent work include the AER-1 conformance suite, the open-source receipt checkers, and the hosted verifier at [zambo.dev/verify](https://zambo.dev/verify), which confirms a receipt against the spec. Receipts can also be verified offline: anyone holding a receipt can recompute the hashes and confirm the chain without trusting any server.

Spec: [draft-zambo-aer1-12](https://www.ietf.org/archive/id/draft-zambo-aer1-12.txt). Project home: [zambo.dev](https://zambo.dev).

**Try it live:** [interactive demo](https://rambozambodotdev.gitlab.io/aer1-hub/try/). Mint a real verifiable receipt in your browser, no install, no signup.


One-line install, no API key, free forever: wrap your agents and the whole [AutoGen](https://microsoft.github.io/autogen/) swarm run emits one hash-chained, offline-verifiable **verifiable workflow receipt** (AER-1, an IETF Internet-Draft, 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. Receipt chain heads can anchor to Nostr and Bitcoin, so anyone can later confirm the record was not changed, without trusting any server.

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-autogen](https://pypi.org/project/aer1-autogen/), AutoGen multi-agent chats
- [aer1-langchain](https://pypi.org/project/aer1-langchain/), LangChain chains, agents, tools, and retrievers via callback handler
- [aer1-llamaindex](https://pypi.org/project/aer1-llamaindex/), LlamaIndex agents via callback handler
- [aer1-smolagents](https://pypi.org/project/aer1-smolagents/), SmolAgents agents via step callbacks
- [aer1-openai-agents](https://pypi.org/project/aer1-openai-agents/), the OpenAI Agents SDK via its TracingProcessor
- [aer1-crewai](https://pypi.org/project/aer1-crewai/), CrewAI crews via the event bus
- [aer1-langgraph](https://pypi.org/project/aer1-langgraph/), LangGraph swarms via callback handler
- [aer1-haystack](https://pypi.org/project/aer1-haystack/), Haystack 2.x pipelines via run wrapping
- [aer1-pydantic](https://pypi.org/project/aer1-pydantic/), Pydantic AI agents via run wrapping and manual tool-call recording
- [aer1-strands](https://pypi.org/project/aer1-strands/), Strands Agents via the typed hook system

See the [AER-1 implementation registry](https://rambozambodotdev.gitlab.io/registry.html) for every implementation.

## Install

```bash
pip install aer1-autogen
```

## Use it

```python
import asyncio
from aer1_autogen import AER1AgentWrapper, AER1AutoGenCollector
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_core.models import (
    ChatCompletionClient, CreateResult, ModelFamily, ModelInfo, RequestUsage,
)

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

class StubClient(ChatCompletionClient):
    """Scripted model client: deterministic replies, zero network, no API key.
    Swap in your own ChatCompletionClient (e.g. OpenAIChatCompletionClient
    with OPENAI_API_KEY set) for live runs. Full version: examples/basic_usage.py"""

    def __init__(self, replies):
        self._replies, self._n = list(replies), 0

    async def create(self, messages, **kw):
        reply = self._replies[min(self._n, len(self._replies) - 1)]
        self._n += 1
        return CreateResult(content=reply, finish_reason="stop",
                            usage=RequestUsage(prompt_tokens=1, completion_tokens=1),
                            cached=False, logprobs=None, thought=None)

    async def create_stream(self, messages, **kw):
        raise NotImplementedError

    async def close(self):
        pass

    def actual_usage(self):
        return RequestUsage(prompt_tokens=0, completion_tokens=0)

    def total_usage(self):
        return RequestUsage(prompt_tokens=0, completion_tokens=0)

    def count_tokens(self, messages, **kw):
        return 1

    def remaining_tokens(self, messages, **kw):
        return 1000

    @property
    def capabilities(self):
        return {"vision": False, "function_calling": False,
                "json_output": False, "structured_output": False}

    @property
    def model_info(self):
        return ModelInfo(vision=False, function_calling=False, json_output=False,
                         structured_output=False, family=ModelFamily.UNKNOWN,
                         context_window=4096)

client = StubClient(["draft: research, build, ship", "review: looks good TERMINATE"])

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)

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

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

### Separate verification verdicts

For honest reporting, use `verify_receipt_verdicts()` instead of a
single boolean. Each dimension gets its own verdict; they are never
conflated:

```python
from aer1_autogen import verify_receipt_verdicts, verdicts_summary

verdicts = verify_receipt_verdicts(receipt)
print(verdicts_summary(verdicts))
# byte_integrity=pass schema_validity=pass issuer_authenticity=not_checked
# evidence_linkage=not_checked anchor_verification=not_checked
# chain_integrity=pass
```

| Dimension | What it checks | Collector behavior |
|---|---|---|
| `byte_integrity` | Merkle root matches the recomputed root | `pass` / `fail` |
| `schema_validity` | All required fields present and well-formed | `pass` / `fail` |
| `issuer_authenticity` | Cryptographic proof of who issued the receipt | Always `not_checked`; the collector does not verify signatures |
| `evidence_linkage` | Upstream evidence bytes match their digests | Always `not_checked`; the collector does not capture upstream evidence |
| `anchor_verification` | Anchor proof is valid and binds the root | `pass` / `fail` / `not_checked` (no proof provided) |
| `chain_integrity` | `seq` values are 1..n in order, no duplicate receipt ids | `pass` / `fail` |

Verdict values are `pass`, `fail`, `not_checked`, and `unavailable`.
`not_checked` is neither a successful check nor evidence of failure.

### What the commitment covers (and what it does not)

The `merkle_root` commits to the ordered list of step `receipt_id`
values. Each step `receipt_hash` commits to that step's tool name,
arguments, and observations. The `output_hash` commits to the final
answer bytes.

What this does **not** cover:

- The commitment does not prove the bytes are unchanged since issuance.
  An attacker able to replace both bytes and digest creates another
  matching pair. Continuity requires comparing against a digest from
  an independently trusted path (a retained receipt, a verified
  signature with a trusted key, or a separately verified anchor).
- A matching hash does not prove the action ran, the data is true, or
  any external outcome occurred. It proves the supplied bytes agree
  with the supplied digest.
- The receipt records the collector's observations. It does not prove
  provider truth, correct reasoning, authorization, or business success.
- Without an anchor, truncation (deleting steps from the end and
  recomputing the root) is not detectable.

### Construction versions

This package uses these exact constructions (see `CONSTRUCTION_VERSIONS`
in the `verdicts` module):

- Merkle tree: `section-8.1-binary-merkle-v1`
- Step digest: `sha256-canonical-json-v1`
- Output commitment: `sha256-utf8-v1`
- Anchor proof: `aer1-anchor-proof-v1`

## 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/<id>.

## License

Apache-2.0
