attest-sdk

Python SDK for the Attest cryptographic agent credential service.

Attest issues RS256-signed JWTs to AI agents. Each token carries:

• att_scope — list of "resource:action" permission strings
• att_chain — ordered delegation lineage (list of JTIs)
• att_depth — delegation depth (0 = root)
• att_intent — SHA-256 hex of the original instruction
• att_tid — task tree UUID shared across the chain
• att_uid — originating human user ID

Install

pip install attest-sdk

# With framework integrations
pip install "attest-sdk[langgraph]"
pip install "attest-sdk[anthropic]"

Basic usage

from attest import AttestClient, IssueParams, DelegateParams

client = AttestClient(
    base_url="http://localhost:8080",
    api_key="your-api-key",
)

# Issue a root credential
token = client.issue(IssueParams(
    agent_id="orchestrator-v1",
    user_id="user-42",
    scope=["research:read", "gmail:send"],
    instruction="Draft and send the quarterly report",
    ttl_seconds=3600,
))
print(token.claims.att_tid)    # task tree UUID
print(token.claims.att_scope)  # ["research:read", "gmail:send"]

# Delegate to a child agent (scope must be a subset)
child = client.delegate(DelegateParams(
    parent_token=token.token,
    child_agent="email-agent-v1",
    child_scope=["gmail:send"],
))

# Offline verify — fetch JWKS once, reuse it
jwks = client.fetch_jwks()
result = client.verify(token.token, jwks=jwks)  # no server call needed
if result.valid:
    print("Issuer:", result.claims.iss)
else:
    print("Invalid:", result.warnings)

# Check revocation
is_revoked = client.check_revoked(token.claims.jti)

# Revoke (cascades to all descendants)
client.revoke(token.claims.jti, revoked_by="orchestrator")

# Audit trail for the whole task tree
chain = client.audit(token.claims.att_tid)
for event in chain.events:
    print(event.event_type, event.agent_id, event.created_at)

Async client

import asyncio
from attest import AsyncAttestClient, IssueParams

async def main():
    async with AsyncAttestClient(api_key="your-api-key") as client:
        token = await client.issue(IssueParams(
            agent_id="async-agent",
            user_id="user-1",
            scope=["files:read"],
            instruction="Read the file",
        ))
        jwks = await client.fetch_jwks()
        result = await client.verify(token.token, jwks=jwks)
        print(result.valid)

asyncio.run(main())

LangGraph integration

from typing import TypedDict
from attest import AttestClient
from attest.integrations.langgraph import AttestState, attest_tool, AttestNodes

client = AttestClient(api_key="your-api-key")

# 1. Extend AttestState with your own fields
class MyState(AttestState):
    messages: list
    instruction: str
    user_id: str

# 2. Issue at graph entry — stores JWT in state["attest_tokens"]["orchestrator-v1"]
graph.add_node("issue", AttestNodes.issue(
    client=client,
    agent_id="orchestrator-v1",
    scope=["research:read", "gmail:send"],
    instruction_key="instruction",
    user_id_key="user_id",
))

# 3. Enforce scope at tool call — raises AttestScopeError if not covered
@attest_tool(scope="gmail:send", agent_id="email-agent-v1")
def send_email(state: MyState, to: str, body: str) -> str:
    ...

# 4. Delegate when spawning a sub-agent
graph.add_node("spawn_email_agent", AttestNodes.delegate(
    client=client,
    parent_agent_id="orchestrator-v1",
    child_agent_id="email-agent-v1",
    child_scope=["gmail:send"],
))

# 5. Revoke at graph teardown
graph.add_node("cleanup", AttestNodes.revoke(
    client=client,
    agent_id="orchestrator-v1",
))

LangGraph HITL approval

For high-risk handoffs, use AttestNodes.gated_delegate(...) so the graph pauses on a LangGraph interrupt() until a human approves:

from attest.integrations.langgraph import AttestNodes

graph.add_node("approve_prod_deploy", AttestNodes.gated_delegate(
    client=client,
    parent_agent_id="orchestrator-v1",
    child_agent_id="deploy-agent",
    child_scope=["deploy:prod"],
    intent="Deploy the approved release to production",
))

If you are using AttestStateGraph, you can also require approval declaratively in scope_map:

graph = AttestStateGraph(
    MyState,
    client=client,
    scope_map={
        "deploy_node": {
            "scope": ["deploy:prod"],
            "require_approval": True,
            "intent": "Deploy the approved release to production",
        },
    },
)

Anthropic SDK integration

from attest import AttestClient
from attest.integrations.anthropic_sdk import AttestSession, attest_tool_anthropic

client = AttestClient(base_url="http://localhost:8080", api_key="your-api-key")

with AttestSession(
    client=client,
    agent_id="claude-orchestrator",
    user_id="usr_alice",
    scope=["web:read", "files:read", "files:write"],
    instruction="Refactor the auth module",
    system_prompt=SYSTEM_PROMPT,   # auto-computes att_ack checksum
) as session:

    @attest_tool_anthropic(scope="web:read")
    def search_docs(query: str) -> str:
        ...

    # Delegate narrower scope to a sub-agent
    child = session.delegate("code-reviewer", ["files:read"])
    run_reviewer(child.token, "src/auth.py")

# Exiting revokes the root credential and all descendants

For approval-gated delegation:

child = session.delegate(
    "deploy-agent",
    ["deploy:prod"],
    require_approval=True,
    intent="Deploy the approved release to production",
)

This path waits for a human approval challenge to resolve before continuing with the narrowed child credential.

For production-grade HITL provenance in Anthropic flows, pass an approval_handler that exchanges the challenge for the actual approved credential:

child = session.delegate(
    "deploy-agent",
    ["deploy:prod"],
    require_approval=True,
    intent="Deploy the approved release to production",
    approval_handler=lambda challenge: get_id_token_from_your_ui(challenge.challenge_id),
)

Without approval_handler, the session can still wait for external approval, but it resumes by issuing a fresh narrowed credential rather than returning the original HITL-stamped token.

Offline verification note

Once you have fetched the JWKS with client.fetch_jwks(), you can verify any token from the same server without making additional network calls:

jwks = client.fetch_jwks()   # one network call

for token_str in incoming_tokens:
    result = client.verify(token_str, jwks=jwks)  # pure local crypto

The public key is stable for the lifetime of the server instance; cache it as long as your process runs.

Verify a signed evidence packet

packet = client.fetch_evidence(token.claims.att_tid)
jwks = client.fetch_jwks(packet.org.id)
verified = client.verify_evidence_packet(packet, jwks=jwks)

print(verified.valid)
print(verified.hash_valid, verified.signature_valid, verified.audit_chain_valid)
print(verified.warnings)

This verifier checks the packet hash, the RS256 packet signature, and the append-only audit hash chain.