Anlyon Python SDK

Governed execution for AI agents. Official Python SDK for Anlyon. Your agent invokes a named action. Anlyon holds the credentials, evaluates policy, waits for approval when required, and makes the production call itself.

Installation

pip install anlyon

Quick start

Define the call once, from an operator key. That is the only place your credential appears:

import os
from anlyon import Client

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

ops.secrets.put("STRIPE_KEY", value=os.environ["STRIPE_KEY"])

ops.actions.create(
    name="refund-order",
    method="POST",
    url_template="https://api.stripe.com/v1/refunds",
    headers={"Authorization": "Bearer {{secret:STRIPE_KEY}}", "Content-Type": "application/x-www-form-urlencoded"},
    requires_approval=True,
)

From then on, this is all your agent sees. No URL, no header, no key:

anlyon = Client(api_key=os.environ["ANLYON_AGENT_KEY"])

# Successful calls expose the canonical response envelope.
result = anlyon.actions.invoke("refund-order", {"charge": "ch_3P9x", "amount": 12000})

if result.data["pendingApproval"]:
    print("Parked for a human:", result.data["approvalId"])
else:
    print("Stripe replied", result.data["responseStatus"])

anlyon.close()

status on an invocation distinguishes failed (the call did not happen, safe to retry) from unknown (it may have reached the destination, so reconcile before retrying). Anlyon never retries an action invocation for you.

These guarantees cover actions routed through Anlyon. A tool your own code calls directly, with a credential your own process holds, is not routed through Anlyon and is not governed by it.

Async

import asyncio
import os
from anlyon import AsyncClient

async def main():
    async with AsyncClient(api_key=os.environ["ANLYON_AGENT_KEY"]) as client:
        result = await client.actions.invoke("refund-order", {"charge": "ch_3P9x", "amount": 12000})
        print(result.data["status"])

asyncio.run(main())

Configuration

from anlyon import Client, RetryConfig

client = Client(
    api_key="anlyon_live_xxx",       # falls back to ANLYON_API_KEY
    base_url="http://localhost:4000",  # falls back to ANLYON_BASE_URL
    timeout=10.0,
    retry=RetryConfig(max_attempts=5),
)

Resources

The governed execution path:

• client.actions: create, import_openapi, list, get, update, delete, invoke, invocations, invocation, versions (hosted execution)
• client.secrets: put, list, get, delete (write-only vault, no method returns a value)
• client.approvals, client.budgets: the controls around an invocation
• client.events: publish, list, get, topics CRUD, subscriptions
• client.analytics, client.logs: usage, delivery analytics, redacted execution history, exports
• client.billing, client.notifications: provider-safe billing state and notification preferences
• client.integrations: Slack and Discord delivery

Write operations accept an idempotency_key so automatic retries can never double-write:

client.actions.invoke("refund-order", {"charge": "ch_1"}, idempotency_key="refund:ord_42")

Error handling

Successful calls return ApiResponse with success, data, pagination, stats, and request_id, matching the HTTP contract and TypeScript SDK. For compatibility with the 0.1 release, indexing, iteration, get(), len(), and equality delegate to response.data. New code should use response.data explicitly. The legacy shortcut is deprecated for removal in the next major release.

Failed requests raise typed exceptions (never return an error envelope):

from anlyon import Client, NotFoundError, RateLimitError

client = Client(api_key="anlyon_live_xxx")
try:
    client.actions.get("act_missing")
except NotFoundError:
    print("Not found")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")

Legacy namespaces

The namespaces below remain supported on the same client and are not part of the governed execution path.

• client.memory: remember, remember_many, recall, forget, ingest, upload_file, summarize, consolidate, jobs, collections
• client.sessions, client.context: persistent conversations and token-budgeted context
• client.cache: lookup, store, list, get, delete, flush (semantic cache)
• client.files: upload, presign, download, list, get, delete
• client.messages: publish, publish_batch, list, get, cancel
• client.queues, client.schedules, client.url_groups, client.dlq: durable delivery controls
• client.workflows: create, list, get, update, delete, trigger, list_runs, get_run, list_run_steps, cancel_run, notify
• client.flow_control: rate and concurrency controls

client = Client(api_key="anlyon_live_xxx")

client.memory.remember(content="The customer prefers email over phone.", idempotency_key="user-123-pref")

result = client.memory.recall(query="how should we contact this customer?")
for r in result.data["results"]:
    print(r["score"], r["content"])

client.messages.publish(url="https://example.com/hook", idempotency_key="order-456")

client.close()