agentcreds-runtime

Runtime agent-identity enforcement built on AgentCreds. Every tool call is authenticated, authorized, and audited - the caller must present a delegation token that is cryptographically authentic, rooted in a credential from a trust anchor you accept, scoped to the tool (and resource) being called, and accompanied by a fresh proof that the caller holds the leaf agent's key.

Two transports share one policy core:

• MCP (interactive, sessionful) - IdentityEnforcer + a FastMCP adapter.
• A2A (agent-to-agent, no MCP server) - A2AVerifier + a self-contained header envelope.

Both enforce the same guarantees: anchor-rooted authority, proof-of-possession, revocation, on-behalf-of principal binding, argument binding, multi-issuer trust, R10 in-token gates (execution-time human approval, carried with the request and relied upon once), human step-up, and tamper-evident audit / ADR.

The two differ only in how the presentation arrives and which core verify call runs. Everything after that is one gate pipeline (PolicyContext._run_gates), in one order - replay, revocation, R10 carried evidence, contextual policy, usage, step-up, R10 one-time commit - pinned by a single order test for both transports. The gate-to-evidence matching itself is the core's (DelegationToken.check_gates_at), so no transport or language re-implements it. Each decision reads the clock once and judges every gate at that instant (PolicyInput.now); the step-up re-reads it after its blocking wait, so a grant that expired while an operator deliberated is expired.

Pure-Python package (the core agentcreds SDK is a compiled PyO3 extension it imports). Install name agentcreds-runtime; import path agentcreds_runtime.

This is the only MCP policy enforcement point. @agentcreds/runtime (Node) is A2A-only by design - R10 gates, step-up approval and the MCP enforcer are not ported, because a second enforcement implementation would have to be kept correct twice and the conformance suite covers this one. A Node agent talks to this PEP over the wire.

Install

pip install agentcreds-runtime            # both transports
pip install "agentcreds-runtime[mcp]"     # + the FastMCP adapter (official MCP SDK)

MCP: per-session enforcement

from agentcreds_runtime import IdentityEnforcer, PolicyConfig

enforcer = IdentityEnforcer(anchor, config=PolicyConfig(max_age_secs=60, audit=my_audit_sink))

challenge_cbor = enforcer.issue_challenge(session_id)   # send to the client
# ... per tool call, with the presentation bytes the client returned:
decision = enforcer.authorize(session_id, tool, arguments, presentation_cbor)
if decision.denied:
    handle(decision.code)            # see "Denial codes" below
chain = enforcer.enforce(session_id, tool, arguments, presentation_cbor)  # or: raises

Holder side: present(token, credential, challenge_cbor, leaf_agent, action=None).

FastMCP adapter

from agentcreds_runtime import IdentityEnforcer
from agentcreds_runtime.fastmcp import guard_tool

enforcer = IdentityEnforcer(anchor)

@mcp.tool()
@guard_tool(enforcer, "tool:search")
async def search(q: str, ctx: Context, agentcreds_presentation: str, agentcreds_chain=None):
    return do_search(q)   # reached only if enforcement passed

A2A: no MCP server

When agent A hands a task directly to agent B, A attaches a self-contained identity header and B verifies it offline. A2A is one-shot, so the sender mints the proof-of-possession challenge with audience set to the receiver.

from agentcreds_runtime import A2AVerifier, make_a2a_header

# Sender (caller):
header = make_a2a_header(token, vc, agent, audience="a2a://orders.example/agent")

# Receiver:
verifier = A2AVerifier(audience="a2a://orders.example/agent", anchor=anchor)
decision = verifier.authorize(header, "tool:search", {"q": "hi"})

verify runs audience-match + anchor-rooted + proof-of-possession; a header minted for another receiver, or outside max_age_secs, is rejected.

Replay protection

A2A is callback-free, so within the freshness window a header could be re-sent verbatim. Each header is made single-use by default (an in-process guard). For more than one receiver replica, pass a shared guard so replays are caught across replicas; to turn it off, set enable_replay_protection=False:

from agentcreds_runtime import A2AVerifier, RedisReplayGuard

verifier = A2AVerifier(audience=me, anchor=anchor)                       # default: on (in-memory)
# verifier = A2AVerifier(..., replay_guard=RedisReplayGuard(redis))      # many replicas
# verifier = A2AVerifier(..., enable_replay_protection=False)            # off

MCP has the same option (IdentityEnforcer(..., replay_guard=...)) but it is off by default: the per-session challenge means legitimate calls reuse byte-identical presentations, so single-use enforcement requires a fresh issue_challenge before each call. A2A senders mint a fresh challenge per message, so it is safe to default on there.

On-behalf-of over A2A (the wire envelope)

There is no session to establish who the human is, so the human's verifiable identity travels with the message and is validated independently by the receiver - restoring the confused-deputy protection. The envelope is a header-name -> value mapping:

from agentcreds_runtime import make_a2a_envelope, A2AVerifier, principal_resolver_from_oidc

# Sender: capability + the human's OIDC token.
envelope = make_a2a_envelope(token, vc, agent, audience=me, principal_token=id_token)
#   {"AgentCreds-A2A": "...", "AgentCreds-A2A-Principal": "AgentCreds-A2A-Principal/1...."}

# Receiver: validate the human via your IdP, then verify the capability.
verifier = A2AVerifier(audience=me, anchor=anchor,
                       principal_resolver=principal_resolver_from_oidc(provider))
decision = verifier.authorize_envelope(envelope, "tool:read_email", {},
                                       resource="mailbox:alice@acme.com/42")

The core then enforces that the capability token's bound principal equals the independently verified human - so a valid token for Bob cannot drive an agent whose authority is bound to Alice.

R10 gates and step-up over A2A

A token whose scope gates a tool on a human decision (Scope.require_approval, Gate.approval_key) is held to it over A2A exactly as over MCP: the receiver refuses the call (approval_required_denied) unless valid, principal-bound, anchor- or approver-key-signed evidence for this exact action is presented, and the evidence is relied upon once (approval_already_consumed on re-use). The evidence travels in the envelope's optional AgentCreds-A2A-Evidence header (base64url of a JSON array of ApprovalEvidence documents); an envelope without it is still well-formed and simply carries no evidence:

evidence = ac.ApprovalEvidence.from_json(grant_json)   # minted by an approver
envelope = make_a2a_envelope(token, vc, agent, audience=me, action=action,
                             canonicalization_profile=CANON_PROFILE_JCS,
                             approval_evidence=[evidence])
# Receiver - the same knobs as IdentityEnforcer:
verifier = A2AVerifier(audience=me, anchor=anchor,
                       approver_directory=directory_cache.current,   # hybrid approval-key
                       consumed_approvals=RedisConsumedApprovals(redis))  # across replicas
decision = verifier.authorize_envelope(envelope, "tool:pay", {"amount": 1})

authorize / authorize_obo / enforce take the same evidence directly as approval_evidence=. The blocking human step-up (approval_policy + approval_client on PolicyConfig) runs over A2A as well, last in the pipeline.

Shared policy (both transports)

Every option below works identically on IdentityEnforcer and A2AVerifier. The shared policy knobs live on a PolicyConfig(...) passed as config=; the trust anchor and transport-specific options stay direct on the constructor.

Denial codes

Decision.code (and the AccessDenied.code raised by enforce) is one of: no_active_challenge, malformed_presentation, possession_failed, not_authorized, credential_invalid, credential_revoked, principal_mismatch, argument_binding_required, untrusted_issuer, replayed_presentation, policy_denied, usage_limit_exceeded, approval_required_denied, approval_already_consumed, canonicalization_profile_mismatch, argument_mismatch, bound_arguments_invalid, store_unavailable, access_denied.

store_unavailable means a stateful backend the decision depends on - the session store, the replay guard or the consumed-approvals ledger - raised instead of answering. It is always a deny with a recorded decision, and unlike the revocation / policy / usage / approval gates it has no fail_open_on_* opt-out.

approval_already_consumed is deliberately distinct from approval_required_denied: the first means the evidence verified and satisfied policy but its reliance unit was already spent (a replay against a valid human approval), the second that the evidence was unsatisfactory. The ADR records the two as separate evaluation / admission verdicts for the same reason - collapsed into one field they are indistinguishable, and they call for opposite responses.

Status

The policy core, the MCP IdentityEnforcer, and the A2A A2AVerifier (including the wire envelope and replay guard) are covered by the test suite, run against the real agentcreds wheel. The FastMCP adapter is intentionally thin over the enforcer. See examples/ for runnable end-to-end flows.