Metadata-Version: 2.4
Name: attest-sdk
Version: 0.1.0b5
Summary: Python SDK for the Attest cryptographic agent credential service
License-Expression: Apache-2.0
Keywords: agent,ai-safety,attest,credentials,delegation,jwt
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security :: Cryptography
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.11
Requires-Dist: httpx>=0.27
Requires-Dist: pyjwt[crypto]>=2.8
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.25; extra == 'anthropic'
Provides-Extra: dev
Requires-Dist: pytest; extra == 'dev'
Requires-Dist: pytest-asyncio; extra == 'dev'
Requires-Dist: respx; extra == 'dev'
Provides-Extra: langgraph
Requires-Dist: langchain-core>=0.3; extra == 'langgraph'
Requires-Dist: langgraph>=0.2; extra == 'langgraph'
Description-Content-Type: text/markdown

# attest-sdk

Python SDK for the [Attest](https://github.com/chudah1/attest-dev) 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

```bash
pip install attest-sdk

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

## Basic usage

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

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

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

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

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

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

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

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

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

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