Anlyon Python SDK

Official Python SDK for Anlyon AI backend primitives.

Installation

pip install anlyon

Quick start

from anlyon import Client

client = Client(api_key="anlyon_live_xxx")  # or set ANLYON_API_KEY

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

# Recall by meaning  successful calls expose the canonical response envelope
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.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.secrets, client.actions, client.approvals, client.budgets agent tools and controls
• 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")