Python SDK

Call Anlyon from Python, sync or async: act for each of your users, invoke governed actions, wait for approvals, read receipts and verify webhooks.

anlyon is the Python client for the Anlyon API, with a synchronous Client and an asynchronous AsyncClient. Your server uses it to let an agent act in your users’ own accounts (Gmail, GitHub, HubSpot, Slack…) without your code ever holding their tokens.

Without it you would build per-user OAuth and token storage, a per-user limit counter, an approval backend and a record of what ran, for every integration. With it, your server names the user once (client.for_user(id)), the agent calls an action by name, and Anlyon adds the user’s token at dispatch, checks the user’s limit, waits for the user’s approval on risky steps and returns a receipt with a grade.

The same calls work with your company’s own accounts: leave out for_user() and the call uses the workspace’s connections and your team’s approvals.

This page documents anlyon 0.8.0. It needs Python 3.10 or later.

Install the SDK

pip install anlyon

For agent frameworks, see LangGraph (with anlyon-langchain), CrewAI and Pydantic AI.

Make your first call for a user

Two keys, two jobs. An operator key declares the action and the rules around it, once, from your deploy script. An app key runs in your server, names the user on every call, and can change none of the rules. See Authentication for creating keys.

import os
from anlyon import Client, EndUserNotConnectedError

# Your server's key. It never reaches the browser or the model.
anlyon = Client(api_key=os.environ["ANLYON_API_KEY"])

# Your server names the user from its own session. The model never does.
user = anlyon.for_user(session.user_id)

try:
    call = user.actions.invoke(
        "open-issue",
        {"title": "Checkout fails on Safari", "body": "Steps to reproduce: ..."},
    ).data
    if call["pendingApproval"]:
        # Parked until this user approves it in your app.
        print("Waiting for the user:", call["approvalId"])
    else:
        print(call["status"], call.get("grade"))  # e.g. succeeded confirmed
except EndUserNotConnectedError as error:
    # The user has not connected GitHub yet. Nothing was sent.
    print("Connect first:", error.connect_url)
finally:
    anlyon.close()

The action open-issue is the one the quickstart declares. Every section below takes one part of this loop further.

Use the async client

AsyncClient has the same resources and methods, awaited. Use it in FastAPI, Pydantic AI or any asyncio app:

import asyncio
import os
from anlyon import AsyncClient


async def main() -> None:
    async with AsyncClient(api_key=os.environ["ANLYON_API_KEY"]) as anlyon:
        user = anlyon.for_user("alice")
        call = (await user.actions.invoke("open-issue", {"title": "Hello", "body": "From my agent."})).data
        if call["pendingApproval"]:
            decided = await user.approvals.wait_for_decision(call["approvalId"], timeout_seconds=600)
            print("Decision:", decided["status"])


asyncio.run(main())

Close a client when you are done: client.close(), await client.close(), or a with / async with block. A client from for_user() shares its parent’s connection pool; closing the bound client leaves the parent open.

Configure the client

import os
from anlyon import Client, RetryConfig

anlyon = Client(
    api_key=os.environ["ANLYON_API_KEY"],  # or set ANLYON_API_KEY and leave it out
    base_url="https://api.anlyon.com",     # or ANLYON_BASE_URL; this is the default
    timeout=30.0,                          # per request, in seconds
    retry=RetryConfig(
        max_attempts=3,
        initial_delay=0.1,
        max_delay=10.0,
        backoff_multiplier=2.0,
        jitter=True,
        retryable_statuses=[429, 500, 502, 503, 504],
    ),
    default_headers={"X-Team": "support"},
)

The SDK retries every request on those conditions, writes included. That is safe because each write that accepts an Idempotency-Key carries one, and the SDK reuses it on its own retries, so a resend is answered from the first attempt’s record. See Make a call safe to repeat.

Check the key at startup

A key with a missing scope fails at the first call that needs it, often in production. A key with too much (for example approvals:decide, which would let your app approve its own requests) fails silently. Check both when your server starts:

from anlyon import Client, ScopeCheckError

anlyon = Client()

try:
    anlyon.auth.require_scopes(
        ["actions:read", "actions:invoke", "end-users:write", "end-users:approvals"],
        forbidden=["approvals:decide", "policies:write", "impact-limits:write", "secrets:write"],
    )
except ScopeCheckError as error:
    raise SystemExit(f"Wrong key. Missing: {error.missing}. Must not hold: {error.forbidden}.")

# Or read the identity yourself.
identity = anlyon.auth.identity().data
print(identity["credentialType"], identity["environmentId"], identity["scopes"], identity["expiresAt"])

auth.identity() calls GET /api/v2/auth/identity and needs no scope. It never returns a secret. Holding a scope is not proof that a call will run: policies, limits, a halted environment and quotas still apply at the call.

Act for one of your users

client.for_user(end_user_id) returns a new client that sends Anlyon-End-User: <end_user_id> on every request. Everything that client does happens for that user:

• An action that uses {{connection:<integration>}} gets that user’s connected account, never the workspace’s.
• Each write counts against the user’s own impact limit when the limit’s scope is end_user.
• An approval policy with decidedBy: "end_user" sends the approval to that user, in your app, not to your team.
• Invocations, approvals and receipts carry the endUserId.

anlyon = Client()

user = anlyon.for_user(session.user_id)  # from your own session
print(user.end_user_id)    # the id
print(anlyon.end_user_id)  # None: the parent stays a workspace client

The rules that keep the user out of the model’s hands:

• The id is your own id for the user: 1 to 128 characters from A-Z a-z 0-9 . _ : @ -. for_user() checks it before anything is sent and raises AnlyonError with code VALIDATION_ERROR otherwise. is_valid_end_user_id(value) checks one yourself.
• The binding is fixed. Calling for_user() with another id on a bound client raises. Build each user’s client from the workspace client.
• Tool schemas never carry a user field. Hand the bound client to your tool adapter, and the model can only choose the action and its input.

Create the app key with Require an end user (endUserRequired: true) when it should never touch the workspace’s own accounts. Every action call it makes without a user is refused with 400 END_USER_REQUIRED (EndUserRequiredError).

Over HTTP, the header is all there is to it:

curl -X POST https://api.anlyon.com/api/v2/actions/open-issue/invoke \
  -H "Authorization: Bearer $ANLYON_API_KEY" \
  -H "Anlyon-End-User: alice" \
  -H "Content-Type: application/json" \
  -d '{ "input": { "title": "Checkout fails on Safari", "body": "Steps to reproduce: ..." } }'

Connect a user’s account

Each user connects their own account once. For an OAuth integration your server makes a connect link, the user signs in at the provider, and your callback confirms the connection for the user in your session. Needs end-users:write.

from anlyon import Client, parse_connect_redirect
from fastapi import Depends, FastAPI, Request
from fastapi.responses import RedirectResponse

app = FastAPI()
anlyon = Client()


@app.post("/connections/github")
def connect_github(user: User = Depends(current_user)):
    # The link works once, for 10 minutes.
    link = anlyon.end_users.create_connect_link(
        end_user_id=user.id,
        integration="github",
        redirect_url="https://app.example.com/connections/callback",
    ).data
    return RedirectResponse(link["url"], status_code=303)


@app.get("/connections/callback")
def callback(request: Request, user: User = Depends(current_user)):
    # Use the user from your session, never one read from the URL.
    result = parse_connect_redirect(str(request.url))
    if result and result.status == "pending_confirmation" and result.flow_id:
        anlyon.end_users.confirm_connection(user.id, result.flow_id)
        # Same thing on a bound client: anlyon.for_user(user.id).connections.confirm(result.flow_id)
    return RedirectResponse("/settings/connections")

• redirect_url must fall under one of the environment’s allowed redirect URLs. Leave it out to use the environment’s default redirect URL. With neither, the call raises RedirectUrlRequiredError. Set both in the console under Integrations → Connect settings.
• The redirect carries anlyon_status (pending_confirmation, connected, declined or error), integration and, while the grant waits, anlyon_flow. parse_connect_redirect() reads them into status, flow_id and integration. It never returns a user id: the URL is not a source of truth.
• Until your server confirms, the connection does not exist. A flow started for another user raises EndUserMismatchError (403 END_USER_MISMATCH), and nothing is connected. That is what stops a forwarded link from attaching someone else’s account to your user.
• At most 10 links per minute per user, and 300 per environment.

For an API-key integration (GitHub, Notion, Jira…) your server can store the user’s own key directly. Anlyon checks it with the provider when the catalogue can, encrypts it and never returns it:

anlyon.end_users.set_connection(user.id, "github", token=key_from_your_form)
anlyon.end_users.set_connection(user.id, "jira", fields={"site": "acme", "email": email, "apiToken": api_token})

connections = anlyon.end_users.list_connections(user.id).data  # end-users:read
anlyon.end_users.delete_connection(user.id, "github")          # revokes, then deletes

A key the provider refuses is not stored (422 CONNECTION_KEY_REJECTED). A connection’s status is active, needs_reconnect or revoked. See Workspace connections and end-user connections.

Handle a user who has not connected

An action call for a user with no working connection raises before anything is sent:

from anlyon import EndUserConnectionRequiredError, EndUserReconnectRequiredError

try:
    anlyon.for_user(user_id).actions.invoke("open-issue", payload)
except EndUserConnectionRequiredError as error:  # both "never connected" and "connect again"
    url = error.connect_url or anlyon.end_users.create_connect_link(
        end_user_id=user_id, integration=error.integration
    ).data["url"]
    action = "reconnect" if isinstance(error, EndUserReconnectRequiredError) else "connect"
    return {"action": action, "integration": error.integration, "url": url}

Both carry integration, end_user_id, connect_url and expires_at. connect_url is a fresh link that works once, for 10 minutes. It is None when the environment has no default redirect URL, or when the integration connects with a key: make a link with create_connect_link() then.

The anlyon-langchain and anlyon-llamaindex tools return {"status": "connect_required", "integration", "connectUrl", "message"} as a tool result instead of raising, so the model can tell the user. The same helper is in the SDK: from anlyon.integrations.action_tools import connect_required_result.

Let the user decide their approvals

An approval policy with decidedBy: "end_user" sends each approval to the user the call was made for. It never appears in your team’s inbox. Your app lists and decides it. Needs end-users:approvals.

# What this user sees in your app: their own approvals, nobody else's.
pending = anlyon.end_users.approvals.list(user.id, status="pending").data
for approval in pending:
    print(approval["id"], approval["title"], approval["payload"])  # the exact request

# The user clicked Approve. Approving runs the parked call.
decision = anlyon.end_users.approvals.decide(
    user.id, approval_id, decision="approve", reason="Looks right"  # or decision="deny"
).data
print(decision["approval"]["status"], decision.get("invocation", {}).get("grade"))

An approval that belongs to another user, or one the team decides, answers 404. Take the user id from your session, never from the request.

Give the browser a short-lived session

The prebuilt React components in @anlyonhq/react run in the browser and must never see your API key. Mint a session token for the signed-in user:

@app.post("/api/anlyon/session")
def anlyon_session(user: User = Depends(current_user)):
    data = anlyon.end_users.create_session(user.id, ttl_seconds=900).data
    return JSONResponse(data, headers={"Cache-Control": "no-store"})

The token (eus_...) lasts 15 minutes by default, 60 to 3600 seconds on request, and never past the key’s own expiry. It can list, connect and disconnect that user’s accounts and, when the key holds end-users:approvals, list and decide their approvals. Nothing else: no action call, no secret, no other user, and it cannot confirm a connection. Needs end-users:write, and an API key. See Components.

List and erase users

# Users with connections or activity in this environment. Needs end-users:read.
cursor = None
while True:
    page = anlyon.end_users.list(cursor=cursor, limit=50)
    for u in page.data:
        print(u["endUserId"], u["connections"], u["invocations"], u["pendingApprovals"])
    cursor = page.pagination["nextCursor"]
    if not cursor:
        break

# A user deletes their account. Needs end-users:write.
erased = anlyon.end_users.erase("alice").data
print(erased["connectionsDeleted"], erased["approvalsCancelled"], erased["countersCleared"], erased["erasedAs"])

Erasing revokes the user’s connections at the provider where it can and deletes them, clears their limit counters and cancels their pending approvals. Invocations, receipts, approvals and audit records are kept with the id replaced by a one-way hash, erasedAs. A client from for_user() cannot list other users.

Define an action

An action is one production call, defined once by an operator key: the URL, the headers, the input schema, and where the credential goes. The agent sends only the input. Needs actions:write.

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

ops.actions.create(
    name="open-issue",
    description="Open one issue in the support repository.",
    method="POST",
    url_template="https://api.github.com/repos/acme/support/issues",
    headers={
        # The connected account of the user the call is made for.
        "Authorization": "Bearer {{connection:github}}",
        "Accept": "application/vnd.github+json",
        "User-Agent": "acme-agent",
        "Content-Type": "application/json",
    },
    body_template={"title": "{{input.title}}", "body": "{{input.body}}"},
    input_schema={
        "type": "object",
        "required": ["title", "body"],
        "additionalProperties": False,
        "properties": {
            "title": {"type": "string", "minLength": 1, "maxLength": 256},
            "body": {"type": "string", "minLength": 1, "maxLength": 20000},
        },
    },
    # Counted before dispatch, so a limit can cap it.
    impact={"dimension": "issues", "amount": "1", "bound": "exact"},
    # Read back after the write. A match grades the receipt "confirmed".
    verify={
        "url": "https://api.github.com/repos/acme/support/issues/{{response.number}}",
        "status": 200,
        "match": [{"path": "title", "equals": "{{input.title}}"}],
    },
)

• Templates take {{input.<field>}}, {{secret:NAME}} for a vaulted secret and {{connection:<integration>}} for a connected account. {{connection:github}} is the user’s own account on a for_user() call, and the workspace’s account otherwise.
• An action with impact, verify or governed=True is governed: it needs a fixed https host, its impact is reserved against limits before dispatch, and every call returns a receipt grade.
• requires_approval=True parks every call for a decision. An approval policy can do the same with conditions and send it to the user.
• An adapter action (adapter={"type": "stripe.refund", ...}) builds the request itself. See Governed effects.
• actions.import_openapi(spec, auth=..., dry_run=True) turns a whole OpenAPI 3.x spec into actions. Read skipped for what it left out. See Any HTTP API.

ops.actions.declare("open-issue", verify=None)     # remove the read-back
ops.actions.declare("open-issue", governed=False)  # remove every declaration
ops.actions.update("open-issue", description="Open one support issue.")

actions = anlyon.actions.list().data          # actions:read
action = anlyon.actions.get("open-issue").data  # by name or act_ id
ops.actions.delete("open-issue")              # archives it; the name stays taken

Invoke an action

user = anlyon.for_user("alice")

call = user.actions.invoke(
    "open-issue",
    {"title": "Checkout fails on Safari", "body": "Steps to reproduce: ..."},
    operation_key=f"ticket:{ticket.id}",  # your business identity for this write
).data

Needs actions:invoke. Every call returns an ApiResponse: .data is the payload, .pagination is set on lists, .request_id holds X-Request-Id, and response["id"] reads from data. The invocation:

{
  "id": "inv_01JABCDEF",
  "actionName": "open-issue",
  "status": "pending_approval",
  "pendingApproval": true,
  "approvalId": "apr_01JABCDEF",
  "responseStatus": null,
  "body": null,
  "effectId": "eff_01JABCDEF",
  "grade": "pending",
  "endUserId": "alice"
}

Other invoke arguments: run_id and parent_span_id attach the call to a run, preview_id binds it to a reviewed preview, depends_on names effects that must already hold an outcome, compensates marks a compensation, and retry_of names the failed invocation this one deliberately retries (with a new idempotency_key).

Read the receipt

A governed action answers with a grade on the invoke response and on single invocation reads. It is absent for an action that is not governed.

# One invocation, later: from a job, a webhook handler or the next turn. actions:read
invocation = user.actions.invocation("inv_01JABCDEF").data
print(invocation["status"], invocation.get("grade"), (invocation.get("decision") or {}).get("explanation"))

# The receipt itself, with its evidence.
effect = user.actions.effect(invocation["effectId"]).data
print(effect["grade"], effect["outcome"], effect["verification"])

# Lists, filtered. Paginated with limit and offset.
anlyon.actions.invocations(action="open-issue", status="unknown", end_user_id="alice")
anlyon.actions.effects(grade="acknowledged", acting_for="end_users")

acting_for="workspace" lists only calls made with the workspace’s own accounts, "end_users" only calls made for a user. effects() also filters by action, outcome, stage and unresolved. See Receipts and grades.

preview = user.actions.preview("open-issue", {"title": title, "body": body}, ttl_seconds=3600).data
print(preview["request"], preview["impact"], preview["expiresAt"])

user.actions.invoke("open-issue", preview["request"], preview_id=preview["id"], operation_key="ticket:42")

# Template action: record what the destination confirms. Sends nothing.
ops.actions.resolve_invocation(
    "inv_...",
    outcome="succeeded",
    evidence="Issue #812 visible in GitHub at 14:02 UTC.",
    external_reference="812",
)

# Governed effect: ask the provider read-only, then record a reasoned resolution.
ops.actions.reconcile_effect("eff_...")  # actions:invoke
ops.actions.resolve_effect("eff_...", outcome="failed", reason="No issue with that title exists.")

Make a call safe to repeat

Networks drop replies. Without protection, the retry that follows opens a second issue or sends a second refund.

• Transport retries are covered for you. The SDK (anlyon 0.8.0, and every release since 0.5.0) sends an Idempotency-Key on every invoke, generates one when you pass none, and reuses it on its own retries. A resend is answered from the recorded invocation and is never executed again.
• Your own retries need your own key. A generated key covers one call to invoke(). When your code can call invoke() twice for one business operation (a queue redelivery, a user double-click, a graph that resumes), pass a key derived from that operation.

user.actions.invoke(
    "open-issue",
    payload,
    idempotency_key=f"ticket:{ticket.id}:open-issue",  # replays the recorded result
    operation_key=f"ticket:{ticket.id}",               # governed: same key + same request = same effect
)

An unknown outcome is never sent again by Anlyon, by any key. See Idempotency and retries.

Store secrets

For a credential that is not a connected account (a Stripe key, a webhook token), use the vault. Values go in and never come back out: there is no method that reads a value. Needs secrets:write.

ops.secrets.put(
    "STRIPE_KEY",
    value=os.environ["STRIPE_KEY"],
    allowed_hosts=["api.stripe.com"],  # never sent anywhere else
    allowed_placements=["header"],     # "url", "header" or "body"
)

secrets = ops.secrets.list().data  # names and metadata only (secrets:read)
ops.secrets.delete("OLD_TOKEN")

An action references it as {{secret:STRIPE_KEY}}. Names match ^[A-Z0-9_]+$. Leaving out allowed_hosts on a rotation keeps the host list, so a rotation never widens where a credential may go; None clears it. See Secrets.

Wait for an approval

When a call parks, you can wait for the decision or come back later. Both read the same approval.

call = user.actions.invoke("open-issue", payload, operation_key=f"ticket:{ticket.id}").data

if call["pendingApproval"]:
    # Long-polls in 55-second hops. Default timeout 300 seconds.
    decided = user.approvals.wait_for_decision(call["approvalId"], timeout_seconds=600)

    if decided["status"] == "pending":
        pass  # Not an error: store call["id"] and read it later.
    else:
        done = user.actions.invocation(call["id"]).data
        print(decided["status"], done["status"], done.get("grade"))

A timeout returns the approval still pending: always branch on status (pending, approved, denied or expired). Reading an approval needs approvals:read. An approval expires after 24 hours by default; see When approval expires or changes.

approval = anlyon.approvals.request(
    title="Send the contract to Acme?",
    description="Draft is ready. Sending is irreversible.",
    payload={"draftId": "doc_9"},
    expires_in_seconds=3600,  # default: the workspace's window, 24 hours unless changed
)
decided = anlyon.approvals.wait_for_decision(approval["id"])

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

inbox = operator.approvals.list(status="pending", decider="team")

# Approving an action's approval also runs the parked call.
result = operator.approvals.approve("apr_...", note="Checked the order.").data
print(result.get("invocation", {}).get("status"), result["selfApproved"])

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 an approver says yes

Route approvals with policies

A policy decides, before dispatch, whether a call needs approval, is approved or is denied, and who decides it. Policies are set by an operator key with policies:write: the app key that a policy governs cannot change it.

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

# Every open-issue call waits for the user it acts for.
ops.approval_policies.create(
    name="The user approves each issue",
    match_kind="action",
    match_action_name="open-issue",
    effect="require_approval",
    decided_by="end_user",
)

# Large refunds need two of your team.
ops.approval_policies.create(
    name="Refunds over $500",
    match_kind="action",
    match_action_name="refund-order",
    condition="input.amount >= 50000",
    effect="require_approval",
    required_approvals=2,
)

# What would decide this call? Read-only. policies:read
evaluation = ops.approval_policies.evaluate(kind="action", action_name="refund-order", payload={"amount": 60000}).data
print(evaluation["decision"]["effect"], evaluation["decision"]["explanation"])

• effect is require_approval, auto_approve or auto_deny. match_kind is action, custom or any; nothing is inferred from the name.
• decided_by is team (the default: the console, Slack or a key with approvals:decide) or end_user. An end_user policy takes one decision, whatever required_approvals says, and a call with no user is always decided by the team.
• condition is a CEL expression over the input. A condition that errors at run time requires approval.
• list, get, update, enable, disable, delete and versions manage policies. plan_file, test_file and apply_file keep them in a YAML file in your repository. See Policy as code and Approvals.

Cap what an agent can do

An impact limit caps how much the governed actions in an environment may change: money, resource mutations, or a unit an action declares (issues, emails). Every applicable limit is reserved immediately before dispatch, so an approval granted earlier cannot push past it. Changing limits needs impact-limits:write, which no key gets by default.

# Each user may open 5 issues a day. One user cannot use up another's.
ops.impact_limits.create(
    name="Issues per user per day",
    dimension="issues",
    period="day",
    amount=5,
    action_name="open-issue",
    scope="end_user",
)

# One shared cap on refunds, in cents.
ops.impact_limits.create(name="Daily refunds", dimension="money", currency="usd", period="day", amount=50_000)

for limit in anlyon.impact_limits.list().data:  # impact-limits:read
    print(limit["name"], limit["scope"]["kind"], limit["consumed"], limit["reserved"], limit["available"])

period is day, month or total. scope is environment (the default: every call draws from one amount) or end_user (each user has the whole amount; a call with no user counts on the shared amount). Scope is fixed once created. update() renames or changes the amount, and archive() stops applying it to new calls. A call over the limit is refused before it is sent and graded refused. See When the limit is hit.

Trace a run

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="support-triage", agent_id="agt_123")

user.actions.invoke("open-issue", payload, run_id=run["id"])

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

trace = anlyon.runs.get(run["id"])         # the run with its span tree
otlp = anlyon.runs.export_trace(run["id"])  # redacted OpenTelemetry Protocol JSON

Run your own code as a hosted function

A hosted function is your own JavaScript or Python, run with no network and no secrets of its own. It changes things only by calling the actions its version allows. A plan run records its writes without sending them, one approval covers that exact list, and an apply run sends it.

ops.functions.create(name="close-stale-issues")  # functions:write
ops.functions.publish("close-stale-issues", language="python", source=source)

user = anlyon.for_user("alice")
payload = {"olderThanDays": 30}

# functions:run and actions:invoke
plan = user.functions.run("close-stale-issues", payload).data

if plan["status"] == "awaiting_approval":
    # Once plan["plan"]["approval"] is approved (by the team, or by the user when every write's policy says end_user):
    applied = user.functions.run("close-stale-issues", payload, mode="apply", plan_run_id=plan["id"]).data
    run = user.functions.runs.get(applied["id"]).data
    print(run["status"], run["receipt"])

A run ends planned, awaiting_approval, succeeded, failed, budget_exceeded or diverged (an apply write that differs from the plan, stopped before it is sent). An apply answers 409 with PLAN_NOT_APPROVED, PLAN_ALREADY_APPLIED or PLAN_INPUT_MISMATCH, and 503 FUNCTION_RUNNER_UNAVAILABLE when the runner cannot be reached. See Run your own code and Hosted functions.

Verify webhooks

Subscription deliveries that Anlyon signs carry an Anlyon-Signature header. Verify it before you trust the body: anyone can POST to your endpoint. 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.

import json
import os

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

app = FastAPI()
receiver = Receiver(os.environ["ANLYON_SIGNING_SECRET"], clock_tolerance=300)  # seconds, the default


@app.post("/webhooks/anlyon")
async def handle_webhook(request: Request):
    body = (await request.body()).decode()  # the raw body, before any JSON parsing
    if not receiver.verify(signature=request.headers.get("anlyon-signature"), body=body):
        raise HTTPException(status_code=401, detail="Invalid signature")
    await process_event(json.loads(body))
    return {"ok": True}

receiver.verify_or_throw(signature=..., body=...) raises SignatureVerificationError instead. A signature older than clock_tolerance fails, which stops a captured delivery from being replayed. See Webhooks for event types and payloads.

Handle errors

Every error the SDK raises is an AnlyonError with code, status, details, request_id and retry_after. Quote request_id to support. The SDK has already retried what is retryable by the time you see an error.

from anlyon import (
    AnlyonError,
    EndUserConnectionRequiredError,
    EndUserRequiredError,
    QuotaExceededError,
    RateLimitError,
    ValidationError,
)

try:
    anlyon.for_user(user_id).actions.invoke("open-issue", payload)
except EndUserConnectionRequiredError as error:
    return show_connect_button(error.integration, error.connect_url)
except EndUserRequiredError:
    raise RuntimeError("Build the client with for_user() on the server.")
except ValidationError as error:
    return bad_input(error.details)
except RateLimitError as error:
    return retry_later(error.retry_after)
except QuotaExceededError:
    return tell_the_user("Monthly allowance used up.")
except AnlyonError as error:
    if error.status == 401:
        raise RuntimeError("Bad or revoked API key.") from error
    if error.status == 404:
        return not_found()
    raise

A gated action never raises for a pending decision: it returns pendingApproval set to True. A refused call (a limit, a policy) is not an exception either: it returns status failed or denied with the reason in error or decision, and a governed receipt graded refused or denied.

Page through lists

actions.invocations, actions.effects, approvals.list and runs.list take limit and offset. end_users.list takes a cursor.

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

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

Related

The package is on PyPI: anlyon.