a2a-spec 0.2.0

a2a-spec
The open specification for testing, validating, and guaranteeing agent-to-agent interactions.

Problem

Multi-agent AI systems are impossible to test reliably with conventional tools.

Here is a concrete example of what goes wrong: you refine a prompt in your triage agent. The agent starts returning issue_type instead of category. Your resolution agent silently receives None for a key field and produces garbage output. No test caught it. Three hours later you find out from a user complaint.

The root causes are systemic:

• LLM outputs are non-deterministic — you cannot diff two live runs and call it a test.
• Agent contracts are implicit — there is no formal definition of what Agent B expects from Agent A.
• Semantic drift is invisible to schemas — a field can stay string while its meaning completely changes.
• CI is too expensive to run live — calling LLMs on every push is slow and costs real money.

Existing tools focus on prompt evaluation or observability. None provide contract testing between agents.

Solution

a2a-spec is a specification, testing, and validation layer for multi-agent systems. You define what one agent expects from another as a YAML spec. You record LLM outputs as snapshots once. You replay them deterministically in CI with zero LLM calls. You detect structural and semantic regressions before they reach production.

Agent A ──[spec]──> Agent B ──[spec]──> Agent C
    │                   │                   │
    └── snapshot ──> replay ──> validate ──> ✓ CI passes

Comparison

How It Works

a2a-spec follows a record-once, replay-forever workflow:

1. Write a spec — a YAML file that defines the contract between two agents: what fields are required, what they mean, and what is forbidden.

2. Record snapshots locally — run a2aspec record with your API keys. It calls your live agents and saves each output as a JSON file.

3. Commit snapshots to git — the JSON files become your test baselines. They are version-controlled, reviewable in PRs, and act as a precise record of what your agents actually produce.

4. Validate in CI — run a2aspec test --replay. It loads the snapshots and validates them against the specs. No API keys. No LLM calls. Runs in milliseconds.

5. Detect drift when prompts change — after modifying a prompt or upgrading a model, run a2aspec record again, then a2aspec diff. The diff engine reports exactly what changed structurally and semantically before you merge.

Quick Start

Install

pip install a2a-spec

With optional features:

pip install a2a-spec[semantic]    # Embedding-based semantic comparison
pip install a2a-spec[langchain]   # LangChain adapter
pip install a2a-spec[dev]         # Testing and linting tools
pip install a2a-spec[all]         # Everything

Initialize a project

a2aspec init --name my-project

This creates:

my-project/
├── a2a-spec.yaml              # Project configuration
└── a2a_spec/
    ├── specs/                  # Agent-to-agent contracts
    │   └── example-spec.yaml
    ├── snapshots/              # Recorded outputs (committed to git!)
    ├── scenarios/              # Test input scenarios
    └── adapters/               # Agent wrappers

Infer a spec from your code

You do not have to write specs by hand. Point a2aspec infer at the consumer function and it reads which fields the consumer actually uses, then types them from the producer's recorded snapshots:

a2aspec infer --producer triage-agent --consumer resolution-agent \
  --source agents/resolution.py:resolution_agent

# Generated by 'a2aspec infer' from agents/resolution.py:resolution_agent and 1 triage-agent snapshot(s).
#   category: required, read at line 15
#   priority: required, read at line 16
spec:
  name: triage-agent-to-resolution-agent
  producer: triage-agent
  consumer: resolution-agent
  structural:
    type: object
    required: [category, priority]
    properties:
      category: {type: string}
      priority: {type: string}

Only fields the consumer reads become the contract, so the producer can change everything else freely. A field read with data.get("priority", "medium") is still required when the producer has always sent it: if the producer renames priority to urgency, the consumer would silently fall back to "medium", and the inferred contract fails instead. Uses the analyzer cannot follow (passing the input to another function, dynamic keys) are reported as warnings.

Define a spec

A spec is a YAML contract between a producer agent and a consumer agent. It defines what the consumer expects across three validation layers:

# a2a_spec/specs/triage-to-resolution.yaml
spec:
  name: triage-to-resolution
  version: "1.0"
  producer: triage-agent
  consumer: resolution-agent
  description: "What the resolution agent expects from triage"

  structural:
    type: object
    required: [category, summary, confidence]
    properties:
      category:
        type: string
        enum: [billing, shipping, product, general]
      summary:
        type: string
        minLength: 10
        maxLength: 500
      confidence:
        type: number
        minimum: 0.0
        maximum: 1.0

  semantic:
    - rule: summary_reflects_input
      description: "Summary must faithfully reflect the customer message"
      method: embedding_similarity
      threshold: 0.5
      source_field: message    # field in the agent's input
      target_field: summary    # field in the agent's output

  policy:
    - rule: no_pii
      description: "Output must not contain PII"
      method: regex
      patterns:
        - '\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b'  # Credit card
        - '\b\d{3}-\d{2}-\d{4}\b'                       # SSN

Record snapshots

a2aspec record  # Calls live agents via adapters, saves outputs to disk

Each recorded output becomes a JSON file committed to git — your deterministic test baseline:

{
  "fingerprint": "a1b2c3d4e5f6",
  "agent_id": "triage-agent",
  "scenario": "billing_overcharge",
  "input": { "message": "I was charged twice for my order #12345" },
  "output": {
    "category": "billing",
    "summary": "Customer reports a duplicate charge on order #12345",
    "confidence": 0.92
  }
}

Test in CI (zero LLM calls)

a2aspec test --replay  # Validates saved snapshots against specs

 PASS  triage-agent :: billing_overcharge    structural ✓  semantic ✓  policy ✓
 PASS  triage-agent :: shipping_delay        structural ✓  semantic ✓  policy ✓
 FAIL  triage-agent :: product_defect        structural ✗  field 'confidence' missing

No API keys needed. No LLM costs. Fully deterministic. Structural and policy checks run in milliseconds; semantic rules need pip install a2a-spec[semantic] and run a small local embedding model (pass --no-semantic to skip them).

Detect semantic drift

After changing a prompt or upgrading a model:

a2aspec record   # Re-record with the new configuration
a2aspec diff     # Compare new outputs against the committed baseline

triage-agent :: billing_overcharge
  summary   MEDIUM  semantic similarity 0.71 (threshold 0.85)
             was: "Customer reports a duplicate charge on order #12345"
             now: "Billing issue: duplicate charge"
  category  NONE    unchanged

record overwrites snapshot files in place; diff compares each one with its committed version (--base REF for another revision). It reports structural changes (fields added/removed/type-changed) and semantic drift (meaning shifted beyond threshold) with severity levels from LOW to CRITICAL, plus scenarios added (NEW) or deleted (GONE) since the baseline. High-severity drift or a deleted snapshot exits with code 2.

Use with coding agents

a2a-spec ships an Agent Skill that teaches coding agents the workflow: run contracts before and after touching an agent, ask before recording live, show diff output, regenerate contracts with infer — and never make a failing contract pass by re-recording, deleting snapshots or loosening the spec. Install it into your agent's skills directory:

a2aspec skill --dir path/to/skills   # writes path/to/skills/a2a-spec/SKILL.md
a2aspec skill                         # or print it

The skill is versioned with the package; re-export it after upgrading.

Core Concepts

Three Validation Layers

Every spec can define three independent layers of validation. Each catches a different class of regression:

You need all three because they are not redundant. A summary field can remain a non-empty string (structural pass) while its meaning silently changes (semantic fail). A field can satisfy both structural and semantic rules while still leaking a credit card number (policy fail).

Concepts Glossary

→ See docs/concepts.md for detailed explanations.

Adapters — Wrap Any Agent

a2a-spec is framework-agnostic. Adapters wrap your agents so the framework can call them during recording and testing.

Plain async functions

from a2a_spec import FunctionAdapter

async def my_triage_agent(input_data: dict) -> dict:
    # Your agent logic (calls OpenAI, Anthropic, local model, etc.)
    return {"category": "billing", "summary": "Customer reports duplicate charge", "confidence": 0.95}

adapter = FunctionAdapter(
    fn=my_triage_agent,
    agent_id="triage-agent",
    version="1.0.0",
    model="gpt-4",
)

HTTP endpoints

from a2a_spec import HTTPAdapter

adapter = HTTPAdapter(
    url="http://localhost:8000/triage",
    agent_id="triage-agent",
    version="1.0.0",
    headers={"Authorization": "Bearer $TOKEN"},
    timeout=30.0,
)

LangChain runnables

from a2a_spec import LangChainAdapter
from langchain_core.runnables import RunnableLambda

chain = RunnableLambda(lambda x: {"category": "billing", "summary": x["message"], "confidence": 0.9})

adapter = LangChainAdapter(
    runnable=chain,
    agent_id="triage-agent",
    version="1.0.0",
)

Custom adapters

from a2a_spec import AgentAdapter, AgentMetadata, AgentResponse

class MyCrewAIAdapter(AgentAdapter):
    def get_metadata(self) -> AgentMetadata:
        return AgentMetadata(agent_id="my-crew-agent", version="1.0")

    async def call(self, input_data: dict) -> AgentResponse:
        result = await my_crew.kickoff(input_data)
        return AgentResponse(output=result.dict())

→ See docs/writing-adapters.md for the full guide.

Pipeline Testing

Test entire multi-agent pipelines as a DAG. a2a-spec validates each agent's output against its spec and checks routing conditions:

pipeline:
  name: customer-support
  agents:
    triage-agent: {}
    billing-agent: {}
    shipping-agent: {}
    resolution-agent: {}
  edges:
    - from: triage-agent
      to: billing-agent
      condition: "output.category == 'billing'"
    - from: triage-agent
      to: shipping-agent
      condition: "output.category == 'shipping'"
    - from: [billing-agent, shipping-agent]
      to: resolution-agent
  test_cases:
    - name: billing_flow
      input: { message: "I was charged twice" }

a2aspec pipeline test pipeline.yaml --mode replay

Routing conditions use a safe AST-based evaluator — no eval() involved.

→ See docs/architecture.md for the pipeline execution model.

Configuration

Project configuration lives in a2a-spec.yaml:

project_name: "my-project"
version: "1.0"

specs_dir: "./a2a_spec/specs"
scenarios_dir: "./a2a_spec/scenarios"

semantic:
  provider: sentence-transformers
  model: all-MiniLM-L6-v2     # Lazy-loaded, only when needed
  enabled: true

storage:
  backend: local
  path: ./a2a_spec/snapshots

ci:
  fail_on_semantic_drift: true
  drift_threshold: 0.15
  replay_mode: exact

Python API

Use a2a-spec programmatically in your existing test suite:

from a2a_spec import load_spec, validate_output, SnapshotStore, ReplayEngine

# Load a spec and validate an output dict against it
spec = load_spec("a2a_spec/specs/triage-to-resolution.yaml")
result = validate_output(
    {"category": "billing", "summary": "Customer charged twice", "confidence": 0.95},
    spec,
)
assert result.passed, result.errors

# Replay a recorded snapshot (no LLM call)
store = SnapshotStore("./a2a_spec/snapshots")
engine = ReplayEngine(store)
output = engine.replay("triage-agent", "billing_overcharge")

# Diff two outputs structurally and semantically
from a2a_spec import DiffEngine
diff = DiffEngine()
results = diff.diff(old_output, new_output, semantic_threshold=0.85)
for r in results:
    print(f"{r.field}: {r.severity} — {r.explanation}")

# Register and run a custom policy validator
from a2a_spec.policy.engine import PolicyEngine
engine = PolicyEngine()
engine.register_validator("no_pii", lambda output, input_data: "ssn" not in str(output))

pytest integration

a2a-spec ships a pytest plugin. Enable it once in pyproject.toml and every spec becomes a test file, with one test per recorded snapshot:

[tool.pytest.ini_options]
a2a_config = "a2a-spec.yaml"
testpaths = ["tests", "a2a_spec/specs"]

$ pytest -v
a2a_spec/specs/triage-to-resolution.yaml::billing_overcharge PASSED
a2a_spec/specs/triage-to-resolution.yaml::shipping_delay FAILED

contract 'triage-to-resolution' violated:
  structural: [<root>] 'priority' is a required property

-k, -x, JUnit XML and IDE test runners work as usual. Run pytest --a2a for a one-off without the ini setting, and --a2a-no-semantic to skip semantic rules. To check live outputs inside your own tests, use the a2a_check fixture:

async def test_triage_live(a2a_check):
    output = await triage_agent({"message": "I was charged twice"})
    a2a_check("triage-to-resolution", output, {"message": "I was charged twice"})

CLI Reference

→ See docs/cli-reference.md for full options and flags.

CI Integration

a2a-spec is designed for CI-first workflows. Record locally with your API keys; test in CI against committed snapshots — no secrets required in CI.

# .github/workflows/a2a-spec.yml
name: Agent Contract Tests
on: [push, pull_request]

jobs:
  spec-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install a2a-spec
      - run: a2aspec test --replay

Output formats:

→ See docs/ci-integration.md for GitHub Actions, Jenkins, and more.

Comparison

Architecture

src/a2a_spec/
├── cli/          # Typer CLI (init, record, test, diff, pipeline)
├── spec/         # Spec schema (Pydantic), YAML loader, JSON Schema validator
├── snapshot/     # Record, store, fingerprint, and replay engine
├── diff/         # Structural (JSON) + semantic (embedding) comparison
├── pipeline/     # DAG builder, topological executor, execution traces
├── adapters/     # Agent wrappers: function, HTTP, LangChain
├── policy/       # Policy engine with regex and custom validators
├── semantic/     # Embedding model interface (sentence-transformers)
├── reporting/    # Console (Rich), Markdown, JUnit XML, GitHub annotations
├── config/       # YAML config loader with Pydantic validation
├── _internal/    # SHA256 hashing, safe expression evaluator, type aliases
└── exceptions.py # Hierarchical error types with actionable messages

→ See docs/architecture.md for the full design.

Examples

The examples/customer_support/ directory contains a complete walkthrough:

• Two agents (triage + resolution) with a2a-spec contract
• YAML spec with structural, semantic, and policy rules
• Pre-recorded snapshot for deterministic replay
• Test scenarios and pytest integration
• Step-by-step README

Documentation

Contributing

Contributions are welcome. See CONTRIBUTING.md for the development setup, check commands, and PR process.

License

Apache 2.0 — see LICENSE for details.