Python SDK

pip install kya-sdk
# or from source
pip install -e "packages/sdk-python[dev]"

from kya_sdk import generate_keypair

public_key, private_key = generate_keypair()
# public_key: base64-encoded Ed25519 public key
# private_key: base64-encoded Ed25519 private key

print(f"Public key: {public_key}")
# Store private_key securely — never expose it

from kya_sdk import canonicalize, hash_payload, sign_payload

payload = {
    "action": "charge_payment",
    "amount": 45,
    "currency": "EUR",
    "agent_id": "agt_01J...",
}

# 1. Canonical JSON
canonical = canonicalize(payload)

# 2. SHA-256 hash
payload_hash = hash_payload(canonical)

# 3. Ed25519 signature
signature = sign_payload(canonical, private_key)

print(payload_hash, signature)

import asyncio
from kya_sdk import AsyncKyaClient, canonicalize, hash_payload, sign_payload

async def main():
    async with AsyncKyaClient(
        base_url="https://api.know-your-agent.dev",
        workspace_id="ws_01J...",
    ) as kya:

        # 1. Request capability
        cap = await kya.request_capability(
            agent_id="agt_01J...",
            action="charge_payment",
            ttl_seconds=300,
        )

        # 2. Sign the payload
        payload = {"action": "charge_payment", "amount": 45}
        canonical = canonicalize(payload)
        payload_hash = hash_payload(canonical)
        signature = sign_payload(canonical, agent_private_key)

        # 3. Verify
        result = await kya.verify(
            agent_id="agt_01J...",
            capability_token=cap.capability_token,
            action="charge_payment",
            payload=payload,
            signature=signature,
            payload_hash=payload_hash,
        )

        if result.decision == "ALLOW":
            print("✅ Action authorized")
        else:
            print(f"❌ Denied: {result.reason_code}")

asyncio.run(main())

from fastapi import FastAPI, HTTPException, Request
from kya_sdk import AsyncKyaClient

app = FastAPI()
kya = AsyncKyaClient(base_url="...", workspace_id="ws_01J...")

@app.post("/charge")
async def charge(request: Request):
    body = await request.json()

    result = await kya.verify(
        agent_id=request.headers.get("x-agent-id"),
        capability_token=body["capability_token"],
        action="charge_payment",
        payload=body["payload"],
        signature=body["signature"],
        payload_hash=body["payload_hash"],
    )

    if result.decision != "ALLOW":
        raise HTTPException(
            status_code=403,
            detail={"code": result.reason_code, "audit_event_id": result.audit_event_id},
        )

    return {"success": True}

from kya_sdk import KyaClient

kya = KyaClient(base_url="...", workspace_id="ws_01J...")

result = kya.verify(
    agent_id="agt_01J...",
    capability_token=capability_token,
    action="charge_payment",
    payload=payload,
    signature=signature,
    payload_hash=payload_hash,
)

print(result.decision)  # ALLOW | DENY