A2A Integration

Integrate ACGP with Agent-to-Agent (A2A) protocols to govern inter-agent requests, tool delegation, and escalation paths.

Integration Goals

A2A governance should enforce: - Identity-aware policy selection per sender/receiver pair - Action-level tripwires before execution - Consistent intervention handling across agent boundaries - End-to-end traceability for delegated operations

Installation

pip install "acgp-sdk[a2a,postgres]"

Basic Adapter Wiring

from acgp import GovernanceSteward, PostgresStateStorage
from acgp.integrations.a2a import A2AGovernanceAdapter

steward = GovernanceSteward.production(
    blueprint_file="blueprint.yaml",
    state_storage=PostgresStateStorage(connection_string="postgresql://runtime/acgp"),
)

governed_a2a = A2AGovernanceAdapter(
    a2a_client=my_a2a_client,
    steward=steward,
)
  1. Receive A2A message and verify sender identity.
  2. Build CognitiveTrace from message intent and parameters.
  3. Evaluate trace with steward before forwarding/delegating.
  4. Apply intervention outcome:
  5. ok/nudge: proceed
  6. escalate: require approval path
  7. block/halt: deny and notify
  8. Persist trace and result for audit.

Multi-Agent Policy Pattern

Use a system-level blueprint for shared constraints and role-specific blueprints for each agent class.

# system blueprint excerpt
tripwires:
  - id: cross_agent_rate_limit
    when:
      hook: tool_call
      tool: delegate_task
    condition: "delegations_per_minute > 200"
    on_fail:
      decision: block
      reason: "Delegation rate exceeded"

Escalation Handling Example

result = steward.evaluate(trace)

if result.intervention in {"ok", "nudge"}:
    response = governed_a2a.forward(message)
elif result.intervention == "escalate":
    approved = request_human_approval(trace, result.message)
    if approved:
        response = governed_a2a.forward(message)
    else:
        response = {"status": "denied", "reason": "approval required"}
else:
    response = {"status": "blocked", "reason": result.message}

if result.flags and result.flags.flagged:
    queue_interagent_review(trace_id=result.trace_id)

Operational Recommendations

  • Enforce mutual authentication between participating agents.
  • Separate internal delegation channels from public API channels.
  • Add replay protection for cross-agent messages.
  • Record both originator and delegate IDs in audit metadata.
  • Monitor intervention rates by agent pair to detect abuse.

Troubleshooting Signals

  • Rising escalate rates usually indicate policy mismatch between roles.
  • Frequent timeouts indicate dependency pressure or oversized traces.
  • Repeated block on one route may indicate stale routing policy.