Build A2A agent trust and authorization framework

Introduction

In production A2A networks, every agent interaction begins with a question the protocol itself cannot answer: should this agent be trusted? Google's A2A protocol defines the transport semantics for task delegation and streaming artifacts, but it deliberately leaves trust establishment, authorization policy, and identity verification to the implementing organization. This design choice is intentional—trust models vary dramatically between an internal microservice mesh and a cross-organizational agent federation. As a senior engineer, your responsibility is to build the trust layer that sits between raw A2A message exchange and your business logic, ensuring that every task delegation, every streamed artifact, and every capability advertisement passes through rigorous identity verification and dynamic authorization checks.

This section builds an end-to-end trust framework that covers three critical concerns: cryptographic identity verification using mutual TLS and signed agent cards, fine-grained authorization policies that control which agents can invoke which capabilities, and a dynamic trust scoring system that adjusts agent privileges based on observed behavior. By the end, you will have a production-grade A2ATrustFramework class that integrates directly with the agent card registry and task delegation pipeline you built in earlier goals.

Key Terminology

• Mutual TLS (mTLS): A transport-layer security protocol where both client and server present X.509 certificates, ensuring bidirectional identity verification before any application data is exchanged.
• Trust Score: A floating-point value between 0.0 and 1.0 representing the framework's confidence in an agent's reliability, updated after every interaction using an exponential moving average.
• Agent Card Signing: The process of attaching a cryptographic signature to an agent's capability advertisement document, enabling receivers to verify the card has not been tampered with and was issued by the claimed agent.
• Delegation Depth: The number of agent-to-agent hops a task has traversed from the original requester, bounded by policy to prevent circular delegation and resource exhaustion.
• Revocation Fingerprint: A truncated SHA-256 hash of an agent's public key stored in a permanent deny-list, preventing re-registration under a different agent ID with the same key material.
• Exponential Moving Average (EMA): A scoring formula where new_score = (1 - α) * old_score + α * outcome that weights recent observations more heavily, creating natural decay for agents with degrading reliability.

Concepts

Trust Score Dynamics and Federation Considerations

The exponential moving average scoring model has specific properties that matter in production. Consider an agent that has been operating successfully for months with a trust score of 0.95. A single failure drops the score to 0.7 * 0.95 + 0.3 * 0.0 = 0.665—still above the default 0.5 threshold. This grace margin is intentional: transient network failures and timeouts should not immediately revoke a trusted agent's access. However, three consecutive failures bring the score to 0.228, well below threshold. This behavior creates an automatic circuit breaker without requiring a separate circuit breaker library.

For cross-organizational agent federation, the trust framework requires additional considerations that affect both the TrustPolicy and AgentIdentity models:

• Certificate Authority Federation: Internal agents share a single CA, but federated agents present certificates from external CAs. The certificate_chain field in AgentIdentity must be validated against a configurable set of trusted root certificates, not a single hardcoded CA. Implement this as a TrustedRootsRegistry that maps organization names to their root CA certificates.

• Trust Score Isolation: A federated agent's trust score should not inherit from its organization's aggregate score. Each agent maintains its own score independently. However, when a new agent from a known organization registers, you may initialize its trust score above the default 0.5 based on the organization's historical reliability—this is a policy decision captured in TrustPolicy.initial_score_override.

• Delegation Chain Verification: When Agent A delegates to Agent B who delegates to Agent C, and Agent C is in a different organization, the full delegation chain must be included in the request headers. The max_delegation_depth field in TrustPolicy prevents unbounded chains, but you should also enforce that each hop in the chain is independently authorized. A delegation from a high-trust agent to a low-trust agent does not elevate the low-trust agent's effective permissions.

• Revocation Propagation: In a federated network, revoking an agent requires notifying all participating organizations. Implement revocation propagation using the same A2A streaming channel—publish a RevocationEvent artifact that all registered agents consume. Agents that fail to acknowledge the revocation within a configurable timeout should themselves be flagged for trust score reduction.

Code Walkthrough

Identity Verification Through Signed Agent Cards

Every agent in an A2A network advertises its capabilities through an agent card—a JSON document served at /.well-known/agent.json. In a zero-trust architecture, the agent card itself becomes the first attack surface. A malicious agent could advertise capabilities it does not possess, impersonate a legitimate agent, or replay a stolen card to intercept task delegations. The solution is cryptographic signing: each agent card must carry a digital signature that the receiving agent can verify against a known certificate authority or a pre-shared public key.

The identity model begins with AgentIdentity, a Pydantic model that binds a cryptographic key pair to an agent's organizational metadata. This identity is not merely a label—it is the root of all authorization decisions. When Agent A delegates a task to Agent B, Agent A first retrieves Agent B's card, verifies the signature against Agent B's published public key, and then checks that the signing certificate chains back to a trusted root. Only after this verification succeeds does the task delegation proceed.

Loading diagram...

This diagram illustrates the full trust establishment flow. Notice that trust verification happens at three distinct layers: the agent card signature verification (application layer), the mutual TLS handshake (transport layer), and the signed task payload (message layer). Defense in depth is not optional in agent networks—a compromise at any single layer must not grant unrestricted access.

Building the Core Trust Framework

The following implementation defines the foundational data models and the A2ATrustFramework class that orchestrates identity verification, authorization checks, and trust score management. The AgentIdentity Pydantic model captures the agent's unique identifier, organizational affiliation, public key material, and the certificate chain used for mutual TLS. The TrustPolicy model defines per-capability authorization rules, including minimum trust score thresholds, allowed organizations, and required authentication levels. The A2ATrustFramework class itself maintains an in-memory registry of verified identities and their associated trust scores, exposing methods for verify_identity, authorize_request, and update_trust_score that the task delegation pipeline calls at each stage.

1from pydantic import BaseModel, Field
2from enum import Enum
3from datetime import datetime, timedelta
4from typing import Optional
5import hashlib
6import hmac
7
8class AuthLevel(str, Enum):
9    NONE = "none"
10    CARD_SIGNED = "card_signed"
11    MTLS = "mtls"
12    MTLS_PLUS_PAYLOAD = "mtls_plus_payload"
13
14class AgentIdentity(BaseModel):
15    agent_id: str
16    organization: str
17    public_key_pem: str
18    certificate_chain: list[str] = Field(default_factory=list)
19    capabilities: list[str] = Field(default_factory=list)
20    issued_at: datetime = Field(default_factory=datetime.utcnow)
21    expires_at: Optional[datetime] = None
22
23    def is_expired(self) -> bool:
24        if self.expires_at is None:
25            return False
26        return datetime.utcnow() > self.expires_at
27
28    def fingerprint(self) -> str:
29        key_bytes = self.public_key_pem.encode("utf-8")
30        return hashlib.sha256(key_bytes).hexdigest()[:16]
31
32class TrustPolicy(BaseModel):
33    capability: str
34    min_trust_score: float = 0.5
35    allowed_organizations: list[str] = Field(default_factory=list)
36    required_auth_level: AuthLevel = AuthLevel.MTLS
37    max_delegation_depth: int = 3
38
39class TrustRecord(BaseModel):
40    identity: AgentIdentity
41    trust_score: float = 0.5
42    successful_interactions: int = 0
43    failed_interactions: int = 0
44    last_interaction: Optional[datetime] = None
45    auth_level_achieved: AuthLevel = AuthLevel.NONE

Authorization and Dynamic Trust Scoring

With identity models defined, the next implementation builds the A2ATrustFramework class that the task delegation pipeline integrates with. This class provides three core methods: register_identity stores a verified agent's identity and initializes its trust record, authorize_request evaluates whether a specific agent can invoke a specific capability based on trust policies, and update_trust_score adjusts an agent's trust score after each interaction using an exponential moving average that weights recent behavior more heavily than historical performance. The scoring formula uses a decay factor of 0.3, meaning each new interaction contributes 30% to the updated score while historical performance retains 70% influence—this prevents a single failure from destroying a long-established trust relationship while still reacting quickly to repeated failures.

1class A2ATrustFramework:
2    def __init__(self):
3        self._identities: dict[str, TrustRecord] = {}
4        self._policies: dict[str, TrustPolicy] = {}
5        self._revoked_fingerprints: set[str] = set()
6
7    def register_identity(self, identity: AgentIdentity) -> bool:
8        if identity.is_expired():
9            return False
10        fp = identity.fingerprint()
11        if fp in self._revoked_fingerprints:
12            return False
13        self._identities[identity.agent_id] = TrustRecord(
14            identity=identity,
15            trust_score=0.5,
16            auth_level_achieved=AuthLevel.CARD_SIGNED,
17        )
18        return True
19
20    def register_policy(self, policy: TrustPolicy) -> None:
21        self._policies[policy.capability] = policy
22
23    def authorize_request(
24        self, agent_id: str, capability: str, delegation_depth: int = 0
25    ) -> tuple[bool, str]:
26        record = self._identities.get(agent_id)
27        if record is None:
28            return False, "Agent identity not registered"
29        if record.identity.is_expired():
30            return False, "Agent identity has expired"
31        policy = self._policies.get(capability)
32        if policy is None:
33            return True, "No policy defined; default allow"
34        if record.trust_score < policy.min_trust_score:
35            return False, f"Trust score {record.trust_score:.2f} below minimum {policy.min_trust_score}"
36        if policy.allowed_organizations and record.identity.organization not in policy.allowed_organizations:
37            return False, f"Organization '{record.identity.organization}' not in allowed list"
38        if record.auth_level_achieved.value < policy.required_auth_level.value:
39            return False, f"Auth level {record.auth_level_achieved} insufficient"
40        if delegation_depth > policy.max_delegation_depth:
41            return False, f"Delegation depth {delegation_depth} exceeds max {policy.max_delegation_depth}"
42        return True, "Authorized"
43
44    def update_trust_score(
45        self, agent_id: str, success: bool, weight: float = 0.3
46    ) -> Optional[float]:
47        record = self._identities.get(agent_id)
48        if record is None:
49            return None
50        outcome = 1.0 if success else 0.0
51        record.trust_score = (1 - weight) * record.trust_score + weight * outcome
52        if success:
53            record.successful_interactions += 1
54        else:
55            record.failed_interactions += 1
56        record.last_interaction = datetime.utcnow()
57        return record.trust_score
58
59    def revoke_agent(self, agent_id: str) -> bool:
60        record = self._identities.pop(agent_id, None)
61        if record is None:
62            return False
63        self._revoked_fingerprints.add(record.identity.fingerprint())
64        return True

Do's and Don'ts

Do's

1. ✓Do enforce all three AuthLevel layers in sequence — CARD_SIGNED → MTLS → MTLS_PLUS_PAYLOAD — before passing a task request to business logic — each layer in the trust framework guards a distinct attack surface (card replay, channel hijacking, message forgery); collapsing them by stopping at AuthLevel.MTLS means a transport-layer session compromise still lets an attacker inject unsigned payloads that the framework never inspects.
2. ✓Do call identity.is_expired() before every verify_identity invocation, even when the agent's TrustRecord.trust_score is high — the AgentIdentity.expires_at field tracks certificate validity independently of behavioral scoring; a high-scoring agent with an expired certificate carries a key whose private material may have been compromised since issuance, and trust score alone cannot detect that.
3. ✓Do set TrustPolicy.max_delegation_depth to 3 or lower for any capability exposed to cross-organizational agents — the framework's default of 3 caps task delegation chains; leaving it unbounded allows a low-trust intermediary to route a request through enough hops that the terminal agent receives the calling principal's authorization context without the scrutiny the framework would apply to a direct invocation.

Don'ts

1. ✗Don't set required_auth_level = AuthLevel.CARD_SIGNED for capabilities exposed to agents outside your organization — a signed agent card proves capability advertisement but not channel authenticity; without AuthLevel.MTLS enforced in TrustPolicy, a man-in-the-middle can replay a cryptographically valid card over an unauthenticated channel and receive task responses the framework was never designed to protect.
2. ✗Don't use identity.fingerprint() as the primary key when addressing TrustRecord entries in the trust registry — the method returns the first 16 hex characters of the SHA256 of public_key_pem, a truncated digest that is susceptible to intentional prefix collisions; use the full agent_id + organization composite from AgentIdentity as the registry key so two agents cannot collide onto the same trust record.
3. ✗Don't initialize all new TrustRecord entries at the uniform default score of 0.5 regardless of the agent's organization — agents from organizations already listed in TrustPolicy.allowed_organizations with a track record of successful interactions deserve a higher cold-start score than first-contact unknowns; treating them identically forces established federation partners through the same trust ramp-up as untested agents, creating unnecessary delegation friction in cross-organizational A2A networks.