Anlyon Python SDK

The production execution runtime 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. Durable retries exist, and they belong to delivery: queues, schedules and workflows.

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.

Supporting state

client = Client(api_key="anlyon_live_xxx")

client.memory.remember(content="The customer prefers email over phone.")

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

client.close()

Async

import asyncio
from anlyon import AsyncClient

async def main():
    async with AsyncClient(api_key="anlyon_live_xxx") as client:
        await client.memory.remember(content="The customer prefers email.")
        result = await client.memory.recall(query="contact preference?")
        print(result.data["results"])

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

• 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.memory remember, remember_many, recall, forget, ingest, upload_file, summarize, consolidate, jobs, collections
• 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.events publish, list, get, topics CRUD, subscriptions
• client.files upload, presign, download, list, get, delete
• client.cache lookup, store, list, get, delete, flush (semantic cache)
• client.sessions, client.context persistent conversations and token-budgeted context
• client.analytics, client.logs usage, delivery analytics, redacted execution history, exports
• client.billing, client.notifications provider-safe billing state and notification preferences
• client.flow_control, client.integrations rate/concurrency controls and Slack/Discord delivery

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

client.memory.remember(content="...", idempotency_key="user-123-pref")
client.messages.publish(url="https://example.com/hook", idempotency_key="order-456")

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.memory.get("mem_missing")
except NotFoundError:
    print("Not found")
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")