Python SDK

Official Python SDK for Anlyon. Define actions, vault secrets, invoke by name, wait for approvals, read outcomes and verify webhooks, with sync and async clients.

The Anlyon Python SDK is a typed client for the Anlyon API with synchronous and asynchronous clients. It covers the governed path: actions, secrets, approvals, runs and webhook verification.

Installation

Install the SDK using pip:

pip install anlyon

Quick Start

Two clients, two keys. The operator key defines the action and puts the credential in the vault. The agent key invokes and holds nothing else.

Synchronous Client

import os
from anlyon import Client

# Operator: run once. The only place the credential appears.
ops = Client(api_key=os.environ["ANLYON_OPERATOR_KEY"])

ops.secrets.put("STRIPE_KEY", value=os.environ["STRIPE_KEY"], allowed_hosts=["api.stripe.com"])

ops.actions.create(
    name="refund-order",
    description="Refund a Stripe charge.",
    method="POST",
    url_template="https://api.stripe.com/v1/refunds",
    headers={"Authorization": "Bearer {{secret:STRIPE_KEY}}", "Content-Type": "application/x-www-form-urlencoded"},
    input_schema={
        "type": "object",
        "properties": {"charge": {"type": "string"}, "amount": {"type": "integer"}},
        "required": ["charge", "amount"],
    },
    requires_approval=True,
)

# Agent: no URL, no header, no key.
anlyon = Client(api_key=os.environ["ANLYON_AGENT_KEY"])

result = anlyon.actions.invoke("refund-order", {"charge": "ch_3P9x", "amount": 12000})

if result.data["pendingApproval"]:
    decided = anlyon.approvals.wait_for_decision(result.data["approvalId"])
    print("Decision:", decided["status"])
else:
    print("Stripe replied", result.data["responseStatus"])

# Close when done
anlyon.close()

Asynchronous Client

import asyncio
import os
from anlyon import AsyncClient

async def main():
    async with AsyncClient(api_key=os.environ["ANLYON_AGENT_KEY"]) as anlyon:
        result = await anlyon.actions.invoke("refund-order", {"charge": "ch_3P9x", "amount": 12000})
        if result.data["pendingApproval"]:
            decided = await anlyon.approvals.wait_for_decision(result.data["approvalId"])
            print("Decision:", decided["status"])
        else:
            print("Stripe replied", result.data["responseStatus"])

asyncio.run(main())

Configuration

Basic Configuration

from anlyon import Client

client = Client(api_key=os.getenv("ANLYON_API_KEY"))

Advanced Configuration

from anlyon import Client, RetryConfig

client = Client(
    api_key=os.getenv("ANLYON_API_KEY"),
    base_url="https://api.anlyon.com",  # Optional: defaults to production
    timeout=60.0,  # Request timeout in seconds
    retry=RetryConfig(
        max_attempts=5,
        initial_delay=0.5,
        max_delay=30.0,
        backoff_multiplier=2.0,
        jitter=True,
    ),
    default_headers={"X-Custom-Header": "value"},
)

The retry block governs the SDK’s own transport retries. The SDK sends a request again after a network error, a timeout, a 429 or a 5xx, up to three attempts by default. That applies to actions.invoke too.

From anlyon 0.5.0 the SDK sends an idempotency key on every invoke and generates one when you do not pass one. It reuses that key across its own transport retries. A resend is answered from the recorded invocation. Earlier versions send a key only when you pass one, so pass one.

A generated key covers one call to invoke. When your own code can call invoke twice for one business operation, pass your own idempotency_key derived from that operation.

This is not an action retry. An invocation whose outcome cannot be confirmed is recorded as unknown, and Anlyon does not send it to the destination again. See Idempotency and retries.

Actions

Define a production call once, then let the agent invoke it by name. The definition carries the URL, headers and input schema, and references secrets as {{secret:NAME}}.

ops.actions.create(
    name="update-ticket-priority",
    method="PATCH",
    url_template="https://api.example.com/tickets/{{input.ticketId}}",
    headers={"Authorization": "Bearer {{secret:HELPDESK_TOKEN}}"},
    input_schema={
        "type": "object",
        "properties": {"ticketId": {"type": "string"}, "priority": {"type": "string"}},
        "required": ["ticketId", "priority"],
    },
    requires_approval=False,
)

result = anlyon.actions.invoke(
    "update-ticket-priority",
    {"ticketId": "T-1042", "priority": "high"},
    idempotency_key="ticket-T-1042-priority-high",
)

print(result.data["status"], result.data["responseStatus"])

actions = anlyon.actions.list()

versions = ops.actions.versions("refund-order")

# One invocation, with the destination's reply and the version that ran
invocation = anlyon.actions.invocation("inv_...")

# Everything waiting for an operator to confirm what happened
unknown = ops.actions.invocations(status="unknown")

# Record what the destination confirms. Needs actions:resolve, and the key
# that made the call can never resolve it.
ops.actions.resolve_invocation(
    "inv_...",
    outcome="succeeded",
    evidence="Refund re_1 visible in the Stripe dashboard at 14:02 UTC.",
    external_reference="re_1",
)

Governed actions

An action can declare how much one invocation changes and how to read the result back. Declaring either one makes the action governed. Its impact is reserved against shared limits before dispatch, and every invocation returns a receipt grade. A governed action needs a fixed https host.

ops.actions.declare(
    "send-email",
    # Counted before dispatch. "emails" is a unit this workspace declares.
    impact={"dimension": "emails", "amount": "count(input.to)", "bound": "exact"},
    # A GET on the action's own host, issued after the write.
    verify={
        "url": "https://api.example.com/v1/messages/{{response.id}}",
        "status": 200,
        "match": [{"path": "status", "equals": "sent"}],
    },
)

# One limit, shared by every agent, session and key in the environment.
ops.impact_limits.create(name="Daily emails", dimension="emails", period="day", amount=500)

ops.actions.declare("send-email", verify=None)     # remove the read-back
ops.actions.declare("send-email", governed=False)  # remove every declaration

result = anlyon.actions.invoke("send-email", {"to": ["a@example.com"], "subject": "Hi"})
print(result.data["grade"])  # "confirmed" once the read-back matches

# One effect, with its evidence
effect = ops.actions.effect(result.data["effectId"]).data
print(effect["grade"], effect["outcome"], effect["verification"])

# Receipts that nothing read back
unread = ops.actions.effects(grade="acknowledged").data

# Receipts waiting on evidence or an operator
open_effects = ops.actions.effects(grade="unknown").data

Secrets

The vault is write-only. Values go in and are never returned. An action uses a secret by name, and the agent that invokes the action never sees it.

ops.secrets.put(
    "STRIPE_KEY",
    value=os.environ["STRIPE_KEY"],
    allowed_hosts=["api.stripe.com"],
    allowed_placements=["header"],
)

# Names and metadata only, never values
secrets = ops.secrets.list()

ops.secrets.delete("OLD_TOKEN")

Names are uppercase (^[A-Z0-9_]+$). Omitting allowed_hosts on a rotation leaves the existing host list alone, so a rotation never widens where a credential may go.

Approvals

Gated actions create approvals on their own. An agent can also ask its human directly. Deciding needs the separate approvals:decide scope, and a key can never decide an approval it requested itself.

approval = anlyon.approvals.request(
    title="Send the contract to Acme?",
    description="Draft is ready. Sending is irreversible.",
    payload={"draftId": "doc_9"},
)

print(approval["id"])  # apr_...

decided = anlyon.approvals.wait_for_decision(approval["id"], timeout_seconds=600)

if decided["status"] == "approved":
    send()
elif decided["status"] == "pending":
    print("Still waiting. Ask again later.")
else:
    print("Not sent:", decided["status"])

operator = Client(api_key=os.environ["ANLYON_OPERATOR_KEY"])

# The inbox
pending = operator.approvals.list(status="pending")

# Approving a gated action also runs the parked invocation
operator.approvals.approve("apr_...", note="Checked the order.")

operator.approvals.deny("apr_...", note="Wrong customer.")

@anlyon.approvals.gate(title="Refund a customer?", timeout_seconds=600)
def refund(order_id: str, amount: float) -> None:
    stripe.Refund.create(...)

refund("ord_42", 129.99)  # runs only after a human approves

Runs

A run is the trace an agent’s work is attributed to. Pass its id when invoking and the invocation, the approval wait and the decision join the same timeline.

run = anlyon.runs.start(name="refund-triage", agent_id="agt_123")

anlyon.actions.invoke("refund-order", {"charge": "ch_3P9x", "amount": 12000}, run_id=run["id"])

anlyon.runs.end(run["id"], status="succeeded", output={"refunded": 1})

# The run with its span tree
trace = anlyon.runs.get(run["id"])

# Redacted OpenTelemetry Protocol JSON
otlp = anlyon.runs.export_trace(run["id"])

Webhook Signature Verification

Subscription deliveries that Anlyon signs carry an Anlyon-Signature header. Verify it before trusting the body. A notification webhook is signed differently, with X-Anlyon-Signature: sha256=<hex> when a webhook secret is set, and the Receiver does not verify that format. See Notifications.

Using the Receiver Class

from anlyon import Receiver, SignatureVerificationError

receiver = Receiver(os.environ["ANLYON_SIGNING_SECRET"])

# Boolean check
if not receiver.verify(signature=request.headers.get("anlyon-signature"), body=request_body):
    return {"error": "Invalid signature"}, 401

# Or raise on failure
try:
    receiver.verify_or_throw(
        signature=request.headers.get("anlyon-signature"),
        body=request_body,
    )
    # Signature is valid, process the webhook
    data = json.loads(request_body)
    process_webhook(data)
except SignatureVerificationError as e:
    # Invalid signature
    return {"error": "Invalid signature"}, 401

FastAPI Webhook Handler

from fastapi import FastAPI, Request, HTTPException
from anlyon import Receiver

app = FastAPI()
receiver = Receiver(os.environ["ANLYON_SIGNING_SECRET"])

@app.post("/webhook")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("anlyon-signature")

    # Verify signature
    if not receiver.verify(signature=signature, body=body.decode()):
        raise HTTPException(status_code=401, detail="Invalid signature")

    # Process the webhook
    data = await request.json()
    await process_webhook(data)

    return {"success": True}

See the Webhooks guide for the event types and payloads.

Error Handling

The SDK provides typed error classes for different error scenarios:

from anlyon import (
    Client,
    AnlyonError,
    AuthenticationError,
    RateLimitError,
    QuotaExceededError,
    ValidationError,
    NotFoundError,
)

anlyon = Client(api_key=os.getenv("ANLYON_AGENT_KEY"))

try:
    result = anlyon.actions.invoke("refund-order", {"charge": "ch_3P9x", "amount": 12000})
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limit exceeded: {e}")
except QuotaExceededError:
    print("Quota exceeded. Wait for the reset or free up room; see the error message.")
except ValidationError as e:
    print(f"Validation error: {e.details}")
except NotFoundError:
    print("Resource not found")
except AnlyonError as e:
    print(f"API error: {e.code} - {e}")

ApprovalDeniedError and ApprovalPendingError are raised by approvals.gate() only. A gated hosted action never raises for a pending decision: it returns pendingApproval set to True and you wait or come back later.

Responses

Every call returns an ApiResponse. response.data is the payload, and response.pagination is present on paginated lists. Indexing (response["id"]) delegates to data, so both spellings work:

approval = anlyon.approvals.request(title="Ship the release?")

approval["id"] == approval.data["id"]

Pagination

Paginated lists such as actions.invocations, actions.effects, approvals.list and runs.list take limit and offset and return pagination alongside data:

page = ops.actions.invocations(status="unknown", limit=50, offset=0)

for invocation in page.data:
    print(invocation["id"], invocation["actionName"])

print(page.pagination)

Resources

• PyPI: https://pypi.org/project/anlyon

Next Steps