Identity, credentials & access

Create server-held node signing identities, sign paths, manage DIDs and credentials, and configure access rules. The server performs signing and stores private signing material; the SDK requests those operations and reads results. This chapter also covers tags, budgets and API tokens.

Start the server with configured credentials. Creating signing keys, enrolling DIDs, deciding tags and writing policies require administrator permission. The access-boundaries example includes config.example.yaml with administrator, node and invoke credentials for an isolated loopback daemon.

Three mechanisms, all server-side

Node identities and credentials have typed SDK wrappers. The access-policy example uses client.transport.request for the REST operations; the examples below follow that form.

Node identities and signed paths

Construct Identity with a client, or use app.identities. Administrators can create, rotate, revoke and sign. Invoke credentials can list identities, read proofs and verify them. The discovery-identity tests exercise the following against a real daemon:

import asyncio

from atfgraph import Agent, Client, ConflictError
from atfgraph.identity import Identity


async def main(url: str) -> None:
    async with Client(url, api_key="discovery-admin") as admin:
        app = Agent(node_id="travel", version="1", client=admin)

        @app.skill()
        async def quote(amount: int) -> dict:
            return {"amount": amount}

        await app.register()
        identities = Identity(admin)
        key = await identities.create("travel")          # or return the existing active key
        assert key.algorithm == "Ed25519" and key.state == "active"

        run = await admin.submit("travel.quote", {"amount": 42}, version="1")
        try:
            await identities.sign(run.id, key.id)          # refused while the path is still active
        except ConflictError as exc:
            assert exc.code == "path_active"

        await run.result(timeout=30)                     # a worker must be serving "travel"
        proof = await identities.sign(run.id, key.id)
        checked = await identities.verify(proof)
        assert checked.valid and checked.trusted and checked.key_state == "active"


asyncio.run(main("http://127.0.0.1:18880"))

The same wrapper is available on an agent. Key lifecycle changes affect proofs as follows:

proofs(traverser_id) lists a path's proofs. A different namespace sees NotFoundError for that traverser. Verification checks the proof rather than creating a new signature.

Credentials

Use app.credentials inside a node or Credentials.for_client(client) in a script. The wrapper covers fourteen credential routes and checks arguments before I/O. Invalid labels, insecure http:// web bases, unsupported kinds, expiry outside one second to one year, and empty tag lists raise ValueError or TypeError locally.

Enrol a node

from atfgraph import Client, Credentials


async def enrol(url: str) -> None:
    async with Client(url) as client:
        credentials = Credentials.for_client(client)
        enrolment = await credentials.enroll("desk")          # administrator, once per registered node
        assert enrolment.active and enrolment.did.startswith("did:key:z6Mk")
        alias = await credentials.did("desk", web_base_url="https://graph.example.test")
        print(alias.also_known_as)                            # ['did:web:graph.example.test:agents:desk']
        resolution = await credentials.resolve(enrolment.did)
        assert resolution.is_local

Register the node before enrolling it. Issuing without enrollment returns did_not_enrolled, as checked by tests/test_credentials_sdk.py.

Issue and verify a custom credential

from atfgraph import Client, Credentials


async def issue(url: str) -> None:
    async with Client(url) as client:
        credentials = Credentials.for_client(client)
        issued = await credentials.issue(
            "desk",
            subject={"id": "urn:example:quote:1", "amount": 1200, "currency": "AUD"},
            type="TravelQuoteCredential",
            expires_in_seconds=3600,
        )
        verified = await credentials.verify(issued.credential)
        assert (verified.valid, verified.trusted, verified.reason) == (True, True, None)

        tampered = dict(issued.credential)
        tampered["credentialSubject"] = {**tampered["credentialSubject"], "amount": 1}
        broken = await credentials.verify(tampered)
        assert (broken.valid, broken.reason) == (False, "invalid_signature")

        later = await credentials.verify(issued.credential, at=issued.expires_at + 600_000)
        assert later.reason == "expired"

The credential tests check two independent verification results:

A credential from another deployment can have a valid signature without being trusted by this namespace. at is an epoch-millisecond verification time. resolve_remote=True permits fetching an unenrolled did:web issuer document.

Certify a finished traverser

issue_execution(traverser_id, issuer_label=...) asks the server to certify a finished execution. The server constructs its subject from recorded status and input, output and error hashes. The credential tests demonstrate:

from atfgraph import Client, Credentials


async def certify(url: str, traverser_id: str) -> None:
    async with Client(url) as client:
        credentials = Credentials.for_client(client)
        receipt = await credentials.issue_execution(traverser_id, issuer_label="desk")
        assert receipt.kind == "execution"
        assert receipt.subject["id"] == f"urn:atfgraph:traverser:{traverser_id}"
        assert receipt.subject["execution"]["status"] == "succeeded"
        again = await credentials.issue_execution(traverser_id, issuer_label="desk")
        assert again.id == receipt.id                           # re-issuing returns the same credential
        listed = await credentials.list(issuer_label="desk", kind="execution")
        assert receipt.id in listed.ids

Credential calls inside a handler carry its task ID and lease token for the worker-authorized issue, get and verify operations. The server restricts that authority: issuing as another node returns 403 issuer_forbidden, certifying an active execution returns traverser_active, and a stale lease returns 409 parent_lease_invalid on reads. Certify a completed child that the handler has already awaited; do not assume the still-running parent can certify itself.

Revoke

credentials.revoke(credential_id, "quote superseded") records revocation and its reason. The document's signature can remain valid while trust changes: tests expect (True, False, "revoked"). list(revoked=False) selects unrevoked credentials; True selects revoked ones.

Tag ledger

Capability tags are subject to a namespace policy. The credential tests configure that policy and request tags as follows:

from atfgraph import Client, Credentials
from atfgraph.credentials import TagRule


async def tags(url: str) -> None:
    async with Client(url) as client:
        credentials = Credentials.for_client(client)
        policy = await credentials.set_tag_policy(               # administrator
            default_mode="manual",
            rules=[TagRule(tags=["travel"], approval="auto", reason="reviewed once")],
            issuer_label="desk",
        )
        assert policy.mode_for("travel") == "auto" and policy.mode_for("pii") == "manual"

        proposed = await credentials.propose_tags("desk", ["Travel", "pii", "travel"])   # node owner
        assert proposed.modes == {"travel": "auto", "pii": "manual"} and proposed.pending == ["pii"]

        decided = await credentials.decide_tags("desk", "approve", tags=["pii"], reason="reviewed")
        assert decided.changed == ["pii"]
        ledger = await credentials.tags(label="desk")
        assert sorted(ledger.approved) == ["pii", "travel"]

The client lowercases and deduplicates tags. Policy modes are auto, manual and forbidden; decisions are approve, reject and revoke. Automatically approved tags refresh a signed tag credential, listed with kind="tag" and verified like other credentials.

Access policies and connector grants

The graph server applies authorization across REST, MCP, A2A and emitter admission. Application handlers do not need to reproduce those policies. In the access-boundaries example, an operator installs rules, then a partner identity exercises permitted and denied calls. Its policy definitions are:

from typing import Any, Dict, List

LABEL = "concierge"
PARTNER_GRANTS = ["traversers.submit", "traversers.get", "traversers.path", "agents.capabilities"]


def policy_bodies(principal: str) -> List[Dict[str, Any]]:
    return [
        {"name": "example-partner-no-refunds", "label": LABEL, "capability": "refund", "principal": principal, "effect": "deny"},
        {
            "name": "example-partner-quote-limits", "label": LABEL, "capability": "quote", "principal": principal, "effect": "allow",
            "constraints": {"currency": {"operator": "==", "value": "AUD"}, "nights": {"operator": "<=", "value": 7}},
        },
        {
            "name": "example-partner-itinerary-rate", "label": LABEL, "capability": "itinerary", "principal": principal, "effect": "allow",
            "rate": {"requests_per_second": 0.001, "burst": 2},
        },
    ]

Install those rules and the partner connector manifest using:

from atfgraph import APIError, Client


async def install(operator: Client, principal: str, bodies: list) -> dict:
    policies = [await operator.transport.request("POST", "/api/v2/access/policies", json_body=body) for body in bodies]
    manifest = {"grants": ["traversers.submit", "traversers.get", "traversers.path", "agents.capabilities"], "enabled": True}
    try:
        current = await operator.transport.request("GET", f"/api/v2/access/connectors/{principal}")
        manifest["expected_revision"] = current["revision"]
    except APIError as exc:
        if exc.status != 404:
            raise
    connector = await operator.transport.request("PUT", f"/api/v2/access/connectors/{principal}", json_body=manifest)
    return {"policies": policies, "connector": connector}

Feature tests start a real daemon with the example configuration and check the partner's outcomes:

Use GET /api/v2/access/connector to inspect the caller's manifest. Preview admission without starting a run using POST /api/v2/access/evaluate with operation, target and input; it returns allowed and reason. Re-running the example installer replaces only policies carrying its own prefix.

Access policies are managed through these REST routes, rather than an access-policy command in the CLI reference.

Tag-scoped policies

Add tags to a policy to match capabilities or nodes carrying any listed tag, in addition to label, capability and principal conditions. Matching reads tags from the label's newest registered version. One rule can therefore deny all pii capabilities to analytics:

curl -s -X POST http://127.0.0.1:18880/api/v2/access/policies -H 'content-type: application/json' \
  -d '{"label":"*","principal":"analytics","effect":"deny","tags":["pii"]}'

Budgets

Budgets check recent consumption when work is admitted. Choose scope principal, label, target or administrator-only namespace; metric tokens, cost_usd or root runs; and window 1h, 24h, 7d or 30d.

A deny budget returns 429 budget_exceeded once recorded consumption reaches the limit. A warn budget reports warnings through access evaluation and budget status without blocking work. Budgets do not stop runs already admitted.

curl -s -X POST http://127.0.0.1:18880/api/v2/budgets -H 'content-type: application/json' \
  -d '{"name":"co-consult daily spend","scope":"label","subject":"co-consult","metric":"cost_usd","limit":25,"window":"24h","action":"deny"}'
curl -s 'http://127.0.0.1:18880/api/v2/budgets/status?target=co-consult.consult'   # allowed + headroom per budget

Manage budgets through GET/POST /api/v2/budgets, and read, update or delete an ID through GET/PATCH/DELETE /api/v2/budgets/{id}. Updates use expected_revision. Per-budget usage reports consumption, remaining allowance, exceeded state and window start. GET /api/v2/budgets/status?target&principal reports matching budgets. Non-administrators can budget only their own principal or registered labels. The Console's Policies page displays window consumption.

API tokens

YAML auth.tokens credentials are checked first and remain valid. Administrators can also create stored API tokens without restarting the server, with expiration, rotation and revocation:

curl -s -X POST http://127.0.0.1:18880/api/v2/auth/tokens -H "authorization: Bearer $ADMIN_TOKEN" \
  -H 'content-type: application/json' -d '{"principal":"ci-runner","roles":["invoke"],"label":"nightly evals","expires_at":1800000000000}'
# {"id":"…","prefix":"atfg_ab1","token":"atfg_…","roles":["invoke"],…}   the plaintext appears in this response only

The server stores a token's SHA-256 rather than its secret and does not log the secret. Listing returns its prefix, principal, roles, label, expiry, revocation and last use. Rotation returns a new secret and revokes the old token after optional grace_seconds. Deletion revokes immediately, producing 401 token_revoked; expired tokens return 401 token_expired. Supply a node token through ATFGRAPH_API_KEY or Client(api_key=...). The Console's Access page shows a new secret only once.

What a worker credential may do

The example node uses admin, worker, allowing both administrative setup and worker enrollment. Current node deployment can also issue scoped invoke, worker tokens for own-label registration, as described in the deployment guide. During a leased handler, the SDK attaches task authority to supported operations:

• Credential issue, get and verify use the running task and constrain the worker to its executing node and subtree (credentials.py).
• Usage, logs and run metadata use that authority through observability.py; see Observability.
• Memory and knowledge resolve session and workflow ownership from the leased task, covered by the SDK integration tests.

Not verified by the examples cited here: exact standalone memory and knowledge outcomes for a token containing only worker. Test direct reads against a real daemon with that token before relying on those boundaries.

Next

• Record task-authorized Usage and logs.
• Publish capabilities with Multi-node discovery.
• Test real credentials using the daemon fixture.