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 action once, from an operator key. That is the one place your credential appears. This example uses the Stripe refund adapter with a Stripe test key. The adapter takes "mode": "live" or "mode": "test" and a key that matches it:

import os
from anlyon import Client

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

ops.secrets.put(
    "STRIPE_TEST_KEY",
    value=os.environ["STRIPE_TEST_KEY"],  # sk_test_... The key must match the action's mode.
    allowed_hosts=["api.stripe.com"],  # Anlyon will not send this key to any other host
)

ops.actions.create(
    name="refund-order",
    requires_approval=True,
    adapter={
        "type": "stripe.refund",
        "config": {"account": "acct_1Nxyz", "mode": "test", "currency": "usd"},
        "credentialSecret": "STRIPE_TEST_KEY",
    },
)

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.
refund = {"paymentIntent": "pi_3Pabc", "amount": 1250, "currency": "usd"}  # $12.50 in minor units
result = anlyon.actions.invoke("refund-order", refund)

if result.data["pendingApproval"]:
    print("Parked for a human:", result.data["approvalId"])
else:
    print("Receipt grade:", result.data["grade"])

anlyon.close()

For an API with no adapter, create() takes url_template and headers with {{secret:NAME}} instead.

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 does not retry an action invocation.

The client does retry its own HTTP request to Anlyon after a network error or a timeout, or when Anlyon answers 429, 500, 502, 503 or 504, up to RetryConfig.max_attempts (3 by default). The SDK sends an idempotency key on every invoke. When you do not pass an idempotency_key, it generates one for that call and sends the same key on each resend. A resend is answered from the recorded invocation and is not executed again.

A generated key covers one call to invoke. Two calls are two keys, and two keys are two invocations. Pass your own idempotency_key when your code may call invoke twice for the same business operation, for example after a restart or from a job that runs again.

A generated key is not returned in the response. If a resend arrives while the original request is still running, the API can answer 409 idempotency_request_in_progress. The SDK does not resend that. The error names the key in its message and in details["idempotency_key"]. Call invoke again with that key to get the result of the original. A call without it is a new invocation.

This covers 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.

Declare impact and verify

Any action can say how much one invocation changes and how to read the result back. Declaring either one makes the action governed.

ops.actions.declare(
    "send-email",
    impact={"dimension": "emails", "amount": "count(input.to)", "bound": "exact"},
    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.
# It counts governed actions routed through Anlyon.
ops.impact_limits.create(name="Daily emails", dimension="emails", period="day", amount=500)

dimension is money, resource_mutations, or a unit you name, such as emails. amount is an expression over the input: an integer, input.<path> or count(input.<path>), with one optional * <integer>. create() and update() take the same impact, verify and governed arguments. None removes a declaration. An argument you leave out keeps what is there. governed=False removes every declaration.

Every governed invocation answers with a receipt grade:

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

for effect in anlyon.actions.effects(grade="unknown").data:
    print(effect["id"], effect["grade"])

An action with no declaration and no adapter is not governed. It creates no effect and returns no grade. Anlyon evaluates the amount you declare. It does not check that the expression describes what the provider will do. A declared action has no provider idempotency, so a lost response stays unknown until a read-back or an operator settles 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:
        refund = {"paymentIntent": "pi_3Pabc", "amount": 1250, "currency": "usd"}
        result = await client.actions.invoke("refund-order", refund)
        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, declare, delete, invoke, invocations, invocation, versions, preview, effects, effect (hosted execution)
• client.impact_limits: list, get, create, update, archive (shared limits in money, resource mutations or a declared unit)
• client.secrets: put, list, get, delete (write-only vault, no method returns a value)
• client.approvals, client.approval_policies: approval requests, decisions and the policies that route them
• client.budgets: monthly caps per agent key on Anlyon operations. They do not cap model tokens or the money an action moves
• client.environment: get, halt, resume
• client.runs: traces of what an agent did

Also on the client:

• client.events: publish, list, get, topics CRUD, subscriptions
• client.analytics, client.logs: usage, delivery analytics, redacted execution history, exports
• client.billing, client.notifications: billing state and notification preferences
• client.integrations: Slack and Discord delivery

Most write operations accept an idempotency_key. The same key and the same request return the recorded result instead of writing twice:

client.actions.invoke("refund-order", refund, 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. That 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")

Earlier namespaces remain available through the API for existing workspaces. See the billing docs.