docs / agent-id / examples

Examples

Runnable recipes. All examples mirror the repo's README and test suite (tests/test_identity.py, test_spend.py, test_rotation.py, test_revocation_feed.py, et al.).

Owner authorizes an agent, publishes a card, verifies scopes

from fg_agent_id import AgentIdentity, OwnerIdentity

owner = OwnerIdentity.generate("acme-corp")
agent = owner.create_agent("acme-buyer", scopes={"converse", "negotiate"})

card = agent.card(endpoints={"http": "https://buyer.example/inbox"})
card.verify()                       # self-verifying: no registry
print(agent.address)                # amp:key:<base58>
print(card.did)                     # did:amp:<base58>

scopes = agent.delegation_chain.verify(agent.address)
assert scopes == frozenset({"converse", "negotiate"})

Scope intersection down a chain

Chains only narrow. The principal grants three scopes, the operator passes on two — the agent ends up with two:

from fg_agent_id import Delegation, DelegationChain

chain = DelegationChain(links=(
    Delegation.grant(principal.keys, principal.address, operator.address,
                     {"converse", "negotiate", "spend"}, ttl_seconds=3600),
    Delegation.grant(operator.keys, operator.address, agent.address,
                     {"converse", "negotiate"}, ttl_seconds=3600),
))
assert chain.verify(agent.address) == frozenset({"converse", "negotiate"})

Audience-bound proof of possession

The verifier issues the challenge, the agent signs it, the verifier consumes it once and checks against its own audience — never the one echoed in the response:

from fg_agent_id import ChallengeStore

store = ChallengeStore()                          # verifier side
challenge = store.issue(audience="https://myapp.example")

response = challenge.respond(agent.keys, agent.address)   # agent side

issued = store.consume(response.challenge_id)     # single use — None the second time
address = response.verify(issued, audience="https://myapp.example")

For multiple worker processes, swap in the shared atomic store:

from fg_agent_id import RedisChallengeStore
store = RedisChallengeStore.from_url("redis://localhost:6379")

Key rotation with a stable name

The identity name is the inception address; rotating changes the key in force, not the name. A registry that learns the chain resolves the name to the current key:

from fg_agent_id import RotatingIdentity, RotationRegistry

identity = RotatingIdentity.create()
rotated = identity.rotate()

assert rotated.identity == identity.identity   # stable name
assert rotated.address != identity.address     # key in force changed

registry = RotationRegistry()
registry.learn(rotated.chain)
assert registry.resolve(identity.identity) == rotated.address

If the registry ever sees two valid rotations at the same sequence (a fork — proven key leak), resolve() fails closed and duplicity_evidence() returns the conflicting records.

Encrypted keys at rest

Passphrase-sealed key files (scrypt + ChaCha20-Poly1305, Python only):

sealed = agent.keys.to_encrypted_bytes(passphrase)     # bytes safe to write to disk

from fg_agent_id import KeyPair
restored = KeyPair.from_encrypted_bytes(sealed, passphrase)

The header is authenticated — a tampered version byte fails to open — and passphrases are NFC-normalized, so the same passphrase typed on macOS and Linux decrypts the same file.

Revocation, distributed by feed

The owner revokes a grant; the feed carries it; a consumer syncs its registry. Every entry re-verifies on admit, gaps abort the delta, and the registry is marked synced only on full success:

from fg_agent_id import RevocationFeed, RevocationRegistry, sync_registry

revocation = owner.revoke(delegation)          # only the issuer can

feed = RevocationFeed()                        # publisher side
feed.append(revocation)

registry = RevocationRegistry()                # consumer side
cursor = sync_registry(registry, feed)

assert registry.is_revoked(delegation)
registry.require_fresh(max_age_seconds=3600)   # fail closed on stale data

Revoked delegations make chain.verify(...) reject any chain that contains them — pass revoked=registry.digests and revoked_keys=registry.revoked_keys.

Spend scope verification

Grant a payment cap in a scope string, compose caps down the chain (min-cap intersection), and check a concrete payment against them:

from decimal import Decimal
from fg_agent_id import SpendAuthority

grant = owner.grant(agent.address, scopes={"pay:usd:tx<=50:total<=200"})
chain = DelegationChain(links=(grant,))

chain.verify(agent.address)                          # 1. verify the chain first
scope = SpendAuthority.verify(chain, "usd",          # 2. then the authority math
                              amount="25.00",
                              spent_so_far=Decimal("150.00"))
# raises SpendScopeError if amount > tx cap, or spent_so_far + amount > total cap

SpendAuthority does authority math only — you supply the ledger of what's already been spent; settlement is out of scope. A link with no spend scope for the asset voids authority entirely.

TypeScript

Everything above has a byte-identical TS counterpart; crypto is async, constructors take options objects, and the public-key accessor is keys.public_:

import {
  KeyPair, addressFromSigningKey, AgentCard,
  Delegation, DelegationChain, ChallengeStore,
} from "@fg/agent-id";

const keys = await KeyPair.generate();
const address = addressFromSigningKey(keys.public_.signing);

const card = await AgentCard.create({ keys, address, name: "my-agent" });
await card.verify();

const grant = await Delegation.grant({
  issuerKeys: keys, issuerAddress: address,
  subjectAddress: worker, scopes: ["read"], ttlSeconds: 3600,
});
const scopes = await new DelegationChain([grant]).verify(worker);

No OwnerIdentity / AgentIdentity wrappers in TS — compose the primitives directly. Encrypted key files and registry snapshots are Python-only; nothing on the wire depends on either.

PreviousAPI reference