agentraas (Python SDK)

Exactly-once execution for AI agents — a thin, dependency-light wrapper around AgentRaaS's SDK-style REST gateway. If your agent code calls Stripe, Twilio, HubSpot, or any other API directly, wrap it with this client so a retry — yours, your framework's, or a flaky network — never becomes a duplicate charge, a duplicate contact, or a duplicate message.

Works against any AgentRaaS deployment: self-hosted (./install.sh, free and unlimited on every tier) or AgentRaaS Cloud.

Install

pip install agentraas

Library mode: no server needed

Wrap any function that has a side effect (charging a card, sending an email, creating a CRM record) so retries and concurrent duplicates run it exactly once. Nothing to deploy: state lives in a local SQLite file by default, or in your own Redis.

from agentraas.local import exactly_once, RedisStore

@exactly_once()                                   # SQLite file, one machine
def charge(customer, amount):
    return stripe.Charge.create(customer=customer, amount=amount).id

charge("cus_1", 4200)   # runs for real
charge("cus_1", 4200)   # retry: returns the cached id, no second charge

# Across machines, with your own idempotency key and window:
@exactly_once(store=RedisStore("redis://localhost:6379/0"),
              key=lambda order: order["id"], ttl=3600)
async def ship(order):
    ...

• If the function raises, nothing is cached and a retry runs it again. That includes the case where the provider did the work but the response was lost, so give the function an idempotency_key parameter and send it to the provider (e.g. Stripe's Idempotency-Key header). The decorator fills it with the same value on every retry:

  @exactly_once()
  def charge(customer, amount, idempotency_key=None):
      return stripe.Charge.create(customer=customer, amount=amount,
                                  idempotency_key=idempotency_key).id
  

• A duplicate that arrives while the first call is still running waits for its result (up to wait=30 seconds, then InFlightError).

• Results must be JSON-serializable (return ids or dicts, not SDK objects).

• RedisStore needs pip install redis.

When you want a dashboard, audit trail, human approval or circuit breaking on top, point the same calls at an AgentRaaS server with the Client below.

Chaos tester: find the calls your agent would run twice

agentraas-chaos sits between your code and an API. The first time each distinct write request (POST/PUT/PATCH/DELETE) arrives, it lets the request execute and then drops the response, the network failure that turns a retry into a double charge. Then it reports every action that executed more than once.

pip install agentraas

# Safe: a built-in mock answers every call, nothing real is touched
agentraas-chaos --mock -- python my_agent.py

# Against a real API: use TEST-mode keys
agentraas-chaos --upstream https://api.stripe.com -- pytest

Your command gets AGENTRAAS_CHAOS_URL (e.g. http://127.0.0.1:8787); use it as the API base URL. Without a command it runs until Ctrl-C, so you can point an n8n workflow or any other tool at it by hand.

agentraas-chaos: 6 write requests, 3 responses dropped after execution

  DUPLICATES: 1 action(s) executed more than once

    2x  POST /v1/charges  (body sha256 623f14d1b00d)
  ...
  Protected by an idempotency key: 2 retries

It exits with code 1 when it finds a duplicate, so it can fail CI. --fault error-after-commit returns a 502 instead of dropping the connection. A retry that repeats an Idempotency-Key it has already seen is counted as protected, not executed again.

In GitHub Actions:

- uses: actions/setup-python@v5
  with: { python-version: "3.12" }
- uses: sumedhchatse/agentraas/src/chaos-action@main
  with:
    run: python my_agent.py

Quickstart

1. Connect an agent from your AgentRaaS dashboard (+ Connect Agent) — this gives you an agentraas_key, plus your org_id and agent_id.
2. Add credentials for the service you're calling (Credentials panel) — AgentRaaS forwards to the real API using those, you never pass a raw upstream API key through this SDK.

import agentraas

client = agentraas.Client(
    agentraas_key="ar_live_...",
    org_id="acme-corp",
    agent_id="billing-bot",
    base_url="http://localhost:13000",  # or your Cloud/self-hosted URL
)

result = client.call("stripe", "charge.create", {"amount": 5000, "currency": "usd"})

Or with dot-notation sugar:

stripe = client.service("stripe")
result = stripe.charge.create({"amount": 5000, "currency": "usd"})

Calling a Custom Action you've registered:

result = client.custom("my-internal-webhook", {"foo": "bar"})

Why org_id / agent_id matter

Omit them and every untagged SDK caller shares one unenforced identity server-side — your per-agent rate limit and audit trail won't tell your traffic apart from anyone else's. Set them once you've connected an agent from the dashboard; it takes two extra kwargs.

Retries are safe

AgentRaaS claims an atomic dedup slot server-side before forwarding anything. If client.call(...) raises because of a timeout or a dropped connection, calling it again with the same payload is safe — it either completes normally or returns the cached result from the call that actually went through. This SDK doesn't retry automatically; your own retry logic (or your agent framework's) can be as aggressive as you want.

Error handling

from agentraas import AgentRaaSError

try:
    client.call("stripe", "charge.create", {"amount": 5000, "currency": "usd"})
except AgentRaaSError as err:
    print(err.status_code, err.req_id, str(err))

License

MIT — see LICENSE. (The AgentRaaS server itself is open-core; see the main repo for its licensing.)