zambo-adk

Zambo verifiable execution receipts for Google ADK (Agent Development Kit) agents.

Attach receipt recording to any ADK agent and every tool execution becomes a hash-chained step in a verifiable workflow receipt (AER-1, Section 8): each step carries the tool name, arguments, result, agent identity, and timestamps, with one Merkle root over the whole run. The hooks only observe: every callback returns None, so ADK executes tools exactly as it would without us. No network calls until you call mint().

Works with LlmAgent and any agent carrying the standard ADK tool callbacks, via per-agent callbacks or a runner-wide plugin.

Install

pip install zambo-adk

Requires google-adk>=1.0 (installed automatically).

Quickstart (zero API keys)

This runs a real ADK agent end to end with a scripted fake model, no keys, no network, and ends at a live receipt URL on zambo.dev:

import asyncio
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types as genai_types

from zambo_adk import AER1ADKCollector, attach_receipts
from zambo_adk.testing import scripted_agent


def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b


async def main():
    collector = AER1ADKCollector(goal="Add 40 and 2")

    agent = scripted_agent(
        "calc", [add],
        script=[("add", {"a": 40, "b": 2})],
        final_text="42",
    )
    attach_receipts(agent, collector)          # line 1: attach

    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name="app", user_id="u1")
    runner = Runner(agent=agent, app_name="app", session_service=session_service)
    message = genai_types.Content(
        parts=[genai_types.Part(text="Add 40 and 2")], role="user"
    )
    async for _event in runner.run_async(
        user_id="u1", session_id=session.id, new_message=message
    ):
        pass                                   # line 2: run

    receipt = collector.finalize(final_answer="42")
    assert collector.verify(receipt) == []     # VALID
    print(collector.mint(receipt))             # line 3: live receipt URL

asyncio.run(main())

Output:

recorded 1 tool call(s), receipt is VALID
live receipt: https://zambo.dev/run/6b3e89e9-1952-4850-aef4-a9534688637c

Swap scripted_agent for your own LlmAgent with a real model and the same three lines give you receipts on every tool call.

Runner-wide plugin

To cover every agent under a Runner, including sub-agents, register the plugin on the App instead of attaching per-agent callbacks:

from google.adk.apps import App
from zambo_adk import AER1ADKCollector, AER1ADKPlugin

collector = AER1ADKCollector(goal="Research the launch plan")
plugin = AER1ADKPlugin(collector)
app = App(name="my_app", root_agent=agent, plugins=[plugin])

Plugins run before agent-level callbacks, so the plugin also sees tool calls that an agent callback short-circuits.

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 tool execution, 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 agent identity:

• seq, receipt_id, tool, receipt_hash, started_at, ended_at, status
• agent: the name of the ADK agent that acted (every step)

Each step receipt_hash is SHA-256 over the canonical JSON of the tool name, arguments, result, and error flag. The hash commits to the content; the receipt stays compact. Tool errors are recorded as steps with status: "error" and the exception captured, so failures are auditable too.

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
• 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

Minting a live Zambo receipt

collector.mint(receipt) submits the finalized AER-1 receipt to zambo.dev for verification and returns the live audit URL (https://zambo.dev/run/<receipt-id>). It uses the free zambo.dev MCP tier: no auth, no API keys. The mint itself is a Zambo tool call, so it carries its own verifiable Zambo receipt.

Notes

• attach_receipts preserves callbacks already on the agent: ours are appended to the existing callback lists.
• The collector never touches the network and never changes what the agents do until mint() is called. A recording failure can never break a run.
• session_id defaults to a fresh UUID per collector; pass your own to correlate receipts across runs.
• ScriptedModel / scripted_agent in zambo_adk.testing are test helpers: scripted tool-call sequences with no model credentials.
• Verified against google-adk 2.11.0.

Spec

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

License

Apache-2.0