A2A Validation¶

Controls which agents can talk to which. Uses security rules to decide.

How it works¶

Agent A calls Agent B
  → System checks the rules: "Is A allowed to talk to B?"
    → Yes? → Message goes through
    → No? → Blocked, error returned, event logged

It’s automatic¶

When you use agent_connection(), validation happens transparently:

from abi_core.common.abi_a2a import agent_connection

# This validates HMAC + OPA policy before sending
async for chunk in agent_connection(my_card, target_card, payload):
    process(chunk)
# Raises PermissionError if denied

Same with AgentInteractionFlow:

workflow = AgentInteractionFlow()
workflow.set_source_card(AGENT_CARD)  # ← Required for validation
workflow.add_node(node)

async for chunk in workflow.run_workflow():
    # Validation happens per-node before execution
    process(chunk)

Validation modes¶

Set via A2A_VALIDATION_MODE in your agent’s config:

Mode	Behavior
`strict`	Deny if Guardian is unavailable or policy says no
`permissive`	Allow if Guardian is unavailable, deny only on explicit policy violation
`disabled`	Skip validation entirely (development only)

The OPA policy¶

File: services/guardian/opa/policies/a2a_access.rego

package a2a_access

default allow := false

# Orchestrator can talk to everyone
allow if {
    input.source_agent.name == "orchestrator"
}

# Planner can talk to orchestrator (bidirectional)
allow if {
    input.source_agent.name == "planner"
    input.target_agent.name == "orchestrator"
}

# Everyone can access semantic layer
allow if {
    input.target_agent.name == "semantic-layer"
}

# Block specific pairs
deny if {
    input.source_agent.name == "untrusted"
    input.target_agent.name == "database"
}

Test a policy¶

curl -X POST http://localhost:8181/v1/data/a2a_access/allow \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "source_agent": {"name": "orchestrator"},
      "target_agent": {"name": "planner"}
    }
  }'
# {"result": true}

Manual validation¶

If you need to check without making the actual call:

from abi_core.security.a2a_access_validator import get_validator

validator = get_validator()
is_allowed, reason = await validator.validate_a2a_access(
    source_agent_card=my_card,
    target_agent_card=target_card,
    message="test",
)

if not is_allowed:
    print(f"Denied: {reason}")

What gets sent to OPA¶

{
  "source_agent": {
    "name": "orchestrator",
    "description": "Coordinates workflows",
    "url": "http://orchestrator:8002"
  },
  "target_agent": {
    "name": "planner",
    "description": "Decomposes tasks",
    "url": "http://planner:11437"
  },
  "communication": {
    "timestamp": "2026-05-11T10:30:00Z",
    "message_preview": "Analyze Q4...",
    "message_length": 150
  },
  "validation_mode": "strict"
}

Troubleshooting¶

Validation always fails — Check OPA is running: curl http://localhost:8181/health

Validation always passes — Check mode isn’t disabled: look at A2A_VALIDATION_MODE in your compose.yaml

PermissionError on every call — Your policy might be missing a rule for your agent pair. Test with curl against OPA directly.

Next step¶

👉 Model Serving