Python SDK

pip install ath-sdk

from ath import ATHGatewayClient, ATHNativeClient, AsyncATHGatewayClient, ATHError

from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization

key = ec.generate_private_key(ec.SECP256R1())
pem = key.private_bytes(
    serialization.Encoding.PEM,
    serialization.PrivateFormat.PKCS8,
    serialization.NoEncryption(),
).decode()

# Use as context manager (auto-closes HTTP connection)
with ATHGatewayClient(
    url="http://gateway.example.com",
    agent_id="https://your-agent.com/.well-known/agent.json",
    private_key=pem,
    key_id="my-key-2024",  # optional
    timeout=60.0,           # optional
) as client:
    ...

# Gateway
doc = client.discover()
for p in doc.supported_providers:
    print(p.provider_id, p.available_scopes)

# Native
doc = client.discover()  # also sets internal api_base
print(doc.app_id, doc.auth.scopes_supported)

reg = client.register(
    developer={"name": "Acme", "id": "acme"},
    providers=[{"provider_id": "github", "scopes": ["read:user", "repo"]}],
    purpose="Code review assistant",
)
print(reg.client_id, reg.agent_status)

auth = client.authorize("github", ["read:user"])
print("Send user to:", auth.authorization_url)
print("Session:", auth.ath_session_id)

token = client.exchange_token("auth-code", "ath_sess_...")
print(token.access_token, token.effective_scopes)

# Gateway
user = client.proxy("github", "GET", "/user")
repos = client.proxy("github", "POST", "/user/repos", {"name": "new-repo"})

# Native
products = client.api("GET", "/products")
order = client.api("POST", "/orders", {"shippingAddress": "..."})

client.revoke()

# After registration — save
client.save_credentials("creds.json")

# In a later session — restore
client.load_credentials("creds.json")

# Or set manually
client.set_credentials(client_id="ath_abc", client_secret="ath_secret_xyz")
client.set_token("ath_tk_...")

import asyncio
from ath import AsyncATHGatewayClient

async def main():
    async with AsyncATHGatewayClient(url=..., agent_id=..., private_key=pem) as client:
        await client.discover()
        await client.register(...)
        auth = await client.authorize("github", ["read:user"])
        token = await client.exchange_token(code, auth.ath_session_id)
        data = await client.proxy("github", "GET", "/user")
        await client.revoke()

asyncio.run(main())

from ath import ATHError

try:
    data = client.proxy("github", "GET", "/user")
except ATHError as e:
    print(f"[{e.code}] {e.message}")
    if e.code == "TOKEN_EXPIRED":
        # re-authorize
        pass
    elif e.code == "NOT_REGISTERED":
        # register first
        pass