a2a-passport

a2a-passport

Portable, pluggable identity for Agent2Agent (A2A) traffic.

One credential layer your agents use to authenticate to each other, that works the same way whether the token comes from a throwaway mock IdP in CI, Auth0, Keycloak, Okta, or any other standards-compliant OAuth2/OIDC provider -- and works whether your agents live on one cloud or are spread across several.

Why this exists

Managed platforms (e.g. Google Cloud's Agent Gateway) now give you mTLS, OAuth handshakes, and A2A-aware observability for free -- if every agent you talk to lives inside that one platform. The moment you need to call an agent that doesn't -- a partner's agent, a different cloud, a different team's Keycloak realm -- you need a credential layer that isn't tied to one vendor's identity model. That's what a2a-passport is: the part of the Agent Gateway idea (scoped per-audience credentials, pluggable verification, zero-token-handling app code) that works across boundaries, not just inside one.

Concretely, that means an agent on AWS, an agent on Azure, and an agent on GCP can all trust each other without agreeing on a shared IdP first:

• AWS agent -> anywhere: integrations.aws.cognito_verifier() lets the receiving side accept Cognito-issued tokens.
• Azure agent -> anywhere: integrations.azure.entra_id_verifier() does the same for Microsoft Entra ID.
• GCP agent -> anywhere: GCP doesn't run a general OAuth2 IdP for workloads, so the pattern is different -- integrations.gcp. GoogleIDTokenProvider mints a Google-signed ID token from whatever identity the workload already has (no client secret at all), and integrations.gcp.google_id_token_verifier() is a JWKSVerifier any AWS- or Azure-hosted agent can point at Google's public certs to accept it.

Each of aws, azure, and gcp is an optional extra -- install only the ones for clouds you're actually federating with.

Design

Two Protocols are the entire extension point:

class TokenProvider(Protocol):
    async def get_token(self, audience: str) -> str: ...

class TokenVerifier(Protocol):
    async def verify(self, token: str, *, audience: str) -> VerifiedClaims: ...

Everything else -- RemoteAgentClient, build_agent_app, PassportAuthMiddleware -- is written against these two interfaces and never touches a raw token or a specific IdP's quirks. Swap the implementation you pass in and nothing else in your agent changes:

A single agent process can hold a JWKSVerifier for its inbound side and multiple OAuth2ClientCredentialsProviders for different outbound audiences -- each pointed at a different IdP if that's how your organization is federated. That's the whole point: no single IdP is baked into the library.

Install

pip install a2a-passport            # core
pip install a2a-passport[otel]      # + OpenTelemetry tracing
pip install a2a-passport[aws]       # + Cognito verifier, Secrets Manager helper
pip install a2a-passport[azure]     # + Entra ID verifier, Key Vault helper
pip install a2a-passport[gcp]       # + Google ID token provider/verifier, Secret Manager helper

From source, for development:

python3 -m venv venv
source venv/bin/activate
pip install -e ".[dev,otel,aws,azure,gcp]"

requirements-lock.txt records the exact dependency versions this repository was last verified against (pip install -r requirements-lock.txt) if you want a byte-for-byte reproducible environment instead of the version ranges in pyproject.toml.

Quickstart

Server -- wrap your agent logic in a FastAPI app with auth enforced:

from a2a_passport import build_agent_app
from a2a_passport.auth.verifiers import JWKSVerifier

app = build_agent_app(
    card=my_agent_card,
    executor=MyAgentExecutor(),
    verifier=JWKSVerifier(
        jwks_uri="https://your-tenant.auth0.com/.well-known/jwks.json",
        issuer="https://your-tenant.auth0.com/",
    ),
    expected_audience="my-agent",
    required_scope="my-agent:invoke",
)
# uvicorn.run(app, host="0.0.0.0", port=8000)

Client -- call another agent with a scoped, auto-refreshed token:

from a2a_passport import OAuth2ClientCredentialsProvider, RemoteAgentClient

token_provider = OAuth2ClientCredentialsProvider(
    token_url="https://your-tenant.auth0.com/oauth/token",
    client_id="my-agent",
    client_secret="...",  # from a secrets manager, not source
)

async with RemoteAgentClient(
    "https://weather-agent.internal", token_provider, audience="weather-agent"
) as client:
    result = await client.send_data({"city": "Paris"})

Cloud integrations: federating AWS, Azure, and GCP agents

Each cloud module is a thin pair of helpers -- a pre-wired TokenVerifier and a way to pull a secret out of that cloud's own secret store -- so the core PassportAuthMiddleware/RemoteAgentClient never need to know which cloud a caller or callee is on.

A GCP-hosted agent calling out to an AWS-hosted agent:

# On the GCP side (caller): mint a Google ID token, no secret needed.
from a2a_passport import RemoteAgentClient
from a2a_passport.integrations.gcp import GoogleIDTokenProvider

async with RemoteAgentClient(
    "https://aws-hosted-agent.example.com", GoogleIDTokenProvider(), audience="aws-hosted-agent"
) as client:
    result = await client.send_data({"question": "..."})

# On the AWS side (callee): accept it by verifying against Google's public JWKS.
from a2a_passport import build_agent_app
from a2a_passport.integrations.gcp import google_id_token_verifier

app = build_agent_app(
    card=my_agent_card,
    executor=MyAgentExecutor(),
    verifier=google_id_token_verifier(),
    expected_audience="aws-hosted-agent",
    required_scope="agent:invoke",
)

The reverse direction (AWS or Azure calling a GCP-hosted agent) uses the same OAuth2ClientCredentialsProvider shown in Quickstart, pointed at Cognito's or Entra ID's token endpoint, with the GCP-hosted agent verifying via integrations.aws.cognito_verifier() or integrations.azure. entra_id_verifier() -- no new mechanism, since Cognito and Entra ID are already standard OAuth2/OIDC providers that JWKSVerifier handles generically.

Pulling secrets from each cloud's own store instead of hardcoding them:

from a2a_passport.integrations.aws import secret_from_secrets_manager
from a2a_passport.integrations.azure import secret_from_key_vault
from a2a_passport.integrations.gcp import secret_from_gcp_secret_manager

client_secret = secret_from_secrets_manager("my-agent/cognito-client-secret")
client_secret = secret_from_key_vault("https://my-vault.vault.azure.net/", "entra-client-secret")
api_key = secret_from_gcp_secret_manager("my-secret", project_id="my-project")

GCP is deliberately the asymmetric one here: Google Cloud's own Agent Gateway already solves this problem natively for agents that live on GCP, so there's no gcp_verifier_for_gcp_tokens-style helper needed on that side -- the value this library adds is specifically the AWS<->GCP and Azure<->GCP legs Agent Gateway doesn't reach.

Repository layout

src/a2a_passport/
  auth/
    base.py          TokenProvider / TokenVerifier protocols, VerifiedClaims
    oauth2.py         OAuth2ClientCredentialsProvider -- generic RFC 6749 client-credentials
    verifiers.py       SharedSecretVerifier (dev), JWKSVerifier (production, any OIDC IdP)
    middleware.py      PassportAuthMiddleware -- enforces a TokenVerifier on every route
  client.py            RemoteAgentClient -- authenticated, retrying client to one agent
  server.py            build_agent_app() -- wires auth + A2A routes into a FastAPI app
  retry.py             default retry/backoff policy
  tracing.py            optional OTel wiring (requires the `otel` extra)
  testing/mock_idp.py   in-process mock OAuth2 IdP, for tests and local demos ONLY
  integrations/
    aws.py               Cognito verifier + Secrets Manager helper (requires `aws` extra)
    azure.py             Entra ID verifier + Key Vault helper (requires `azure` extra)
    gcp.py               Google ID token provider/verifier + Secret Manager helper (requires `gcp` extra)

examples/demo_agents/  three agents (weather, translator, orchestrator) proving
                        a real multi-hop A2A chain with per-hop scoped credentials
examples/run_demo.py   starts all four services and drives the chain end-to-end

tests/                 unit tests (auth, middleware) + an in-process conformance suite

Running the demo

python3 examples/run_demo.py

Starts a mock IdP and three agents as separate processes, confirms an unauthenticated call to the orchestrator is rejected (401), then drives a plan_trip request that fans out to the weather and translator agents with independently-scoped tokens for each.

Tests

pytest tests/ -v

Runs fully in-process via httpx.ASGITransport -- no real sockets, fast enough for CI on every commit.

Known rough edges (honest, not hidden)

• A benign RuntimeWarning appears when running the mock IdP via python -m a2a_passport.testing.mock_idp (as examples/run_demo.py does): Python warns that the module was already present in sys.modules because a2a_passport.testing.__init__ re-exports names from it. Harmless -- the server starts and serves correctly either way -- but noisy in logs.
• Task store is in-memory (InMemoryTaskStore) by default -- fine for a single instance, not for multi-instance production deployments. Pass your own task_store= to build_agent_app for that case.
• Cloud integrations are unit-tested against mocks, not real accounts. cognito_verifier, entra_id_verifier, and the GCP ID token provider/verifier are covered by tests that mock boto3/azure-identity/ google-auth at the boundary -- correct URL construction and control flow are verified, but none of them has been run against a real Cognito user pool, Entra ID tenant, or GCP service account yet. Treat them as a strong starting point, not a substitute for testing against your actual tenant before production use.

What this is not (yet)

Contributions welcome on any of these -- see CONTRIBUTING.md:

• No agent registry/discovery. Agents are addressed by URL today. A registry service is a natural extension point but out of scope for v0.1.
• No persistent task store shipped. build_agent_app defaults to a2a-sdk's InMemoryTaskStore; pass your own for multi-instance deployments.
• No streaming (SSE). All agents use AgentCapabilities(streaming=False).
• Mock IdP is HS256/shared-secret, on purpose. It's a test fixture, not a reference IdP implementation -- never point production traffic at it.

Security

Found a vulnerability? Please see SECURITY.md -- do not open a public issue.

License

Apache 2.0 -- see LICENSE.