Metadata-Version: 2.5
Name: agentcreds-runtime
Version: 0.2.0
Summary: Runtime AgentCreds agent-identity enforcement for tool calls - MCP middleware and agent-to-agent (A2A) verification
Project-URL: Repository, https://github.com/agentcreds/agentcreds
License: Apache-2.0
License-File: LICENSE
Classifier: Framework :: AsyncIO
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Security
Requires-Python: >=3.8
Requires-Dist: agentcreds<0.3,>=0.2.0
Provides-Extra: cedar
Requires-Dist: cedarpy>=3.0; extra == 'cedar'
Provides-Extra: mcp
Requires-Dist: mcp<2,>=1.2.0; extra == 'mcp'
Provides-Extra: redis
Requires-Dist: redis>=5; extra == 'redis'
Provides-Extra: test
Requires-Dist: pytest>=7.0; extra == 'test'
Description-Content-Type: text/markdown

# agentcreds-runtime

Runtime **agent-identity enforcement** built on
[AgentCreds](https://github.com/agentcreds/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

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

## MCP: per-session enforcement

```python
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

```python
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.

```python
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`:

```python
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:

```python
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:

```python
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.

| Feature | How |
|---|---|
| **Revocation** | `config=PolicyConfig(revocation_check=...)` (a callable; `revocation_check_from_list(list, anchor)` for a list held in process, or `RevocationListCache(anchor_for=...).check` to fetch each issuer's signed list on a TTL with a signed-age bound, on by default at 3600 s). Denied -> `credential_revoked`. Fail-closed by default. |
| **Multi-issuer** | `anchor_for=` (a resolver; or `anchor_resolver_from_registry(registry)`) instead of a single `anchor`. Untrusted issuer -> `untrusted_issuer`. |
| **Argument binding** | Sender binds with `make_a2a_header(..., action=...)` / `present(..., action=...)`; receiver sets `config=PolicyConfig(require_argument_binding=True)`. Tampered args -> `possession_failed`; unbound when required -> `argument_binding_required`. |
| **On-behalf-of** | MCP: `enforcer.bind_principal(session_id, human_did)`. A2A: the principal envelope above. Mismatch -> `principal_mismatch`. |
| **R10 in-token gates** | `recognized_gate_kinds=`, `approver_directory=`, `consumed_approvals=` on either constructor. Evidence: MCP via the `agentcreds_approval_evidence` tool argument, A2A via the `AgentCreds-A2A-Evidence` header or `approval_evidence=`. Missing/invalid -> `approval_required_denied`; re-use -> `approval_already_consumed`. |
| **Human step-up** | `config=PolicyConfig(approval_policy=..., approval_client=...)` - a flagged call blocks until an operator approves; the returned evidence is verified when it arrives. Denied/timeout -> `approval_required_denied`. |
| **Audit / ADR** | `config=PolicyConfig(audit=..., adr_sink=..., adr_stream=...)` - a structured `AuthzDecision` for every allow *and* deny, carrying the security signal, the correlation `vc_id`, and **who answers** (`accountable_party` + `accountability_source`, read from the credential). With a stream, records fold into a tamper-evident hash-chain (`sign_checkpoint` -> `AdrStream.replay`). |
| **Tracing an effect** | `Decision.record_id` is the ADR's per-call id; `guard_tool` hands it to a handler declaring `agentcreds_record_id`. Log it beside whatever the call changes. An ADR proves authority was *checked*, not that the tool ran - and `vc_id` joins on the credential, which every call shares, so this id is what makes the pairing exact. |
| **Declared canonicalization** | `config=PolicyConfig(canonicalization_profile=..., require_canonicalization_profile=...)`; holders declare theirs via the `agentcreds_canon_profile` argument. A disagreeing profile -> `canonicalization_profile_mismatch` rather than `possession_failed`, so an interop defect is distinguishable from altered arguments. Both refuse - the distinction is diagnostic, which is why the declaration may be unauthenticated. |
| **Freshness by consequence** | `config=PolicyConfig(max_age_by_autonomy={3: 5, 0: 300})` maps the credential's `autonomy_level` to a tighter `max_age_secs`. Only ever narrows: an entry longer than the global bound is clamped. |
| **No blind retry** | `@idempotent(store)` falls back to `agentcreds_record_id` when the caller supplies no key, so duplicate suppression does not depend on client cooperation. Not a substitute for reconciling an indeterminate post-dispatch outcome. |
| **Horizontal scale** | MCP: `session_store=RedisSessionStore(redis)`. A2A: `replay_guard=RedisReplayGuard(redis)`. Every Redis/in-memory store pair is one thin wrapper over the `TtlKv` primitive. A store that raises on the authorization path denies with `store_unavailable` - there is no fail-open for stores. |

## 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/](examples/) for runnable end-to-end flows.
