AgenTrux — Usage Examples

Practical examples showing common patterns for agent-to-agent communication.

Base URL: https://api.agentrux.com

1. Basic: Redeem → Send → Receive

The simplest flow. One agent sends, another receives.

Step 1: Redeem your Activation Code

curl -X POST https://api.agentrux.com/auth/redeem-activation-code \
  -H "Content-Type: application/json" \
  -d '{"code": "act_YOUR_ACTIVATION_CODE"}'

Response:

{
  "client_id": "crd_019d0a77-5449-7a41-8f0a-6062aa283e2e",
  "client_secret": "aks_xxxxxxxx",
  "script_id": "scr_...",
  "issued_at": "2026-03-17T12:00:00+00:00"
}

Save client_id (crd_) and client_secret (aks_) permanently. aks_ is shown only once and you cannot redeem the same code twice.

Step 2: Get an access token

curl -X POST https://api.agentrux.com/oauth/token \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d 'grant_type=client_credentials&client_id=crd_019d0a77-...&client_secret=aks_xxxxxxxx'

Step 3: Send an event

curl -X POST https://api.agentrux.com/topics/019d0a77-52d5-.../events \
  -H "Authorization: Bearer eyJhbG..." \
  -H "Content-Type: application/json" \
  -d '{
    "event_type": "task.assigned",
    "payload": {
      "task": "Analyze sales data for Q1",
      "priority": "high",
      "deadline": "2026-03-25T00:00:00Z"
    }
  }'

Step 4: Receive events

curl https://api.agentrux.com/topics/019d0a77-52d5-.../events \
  -H "Authorization: Bearer eyJhbG..."

2. Python SDK — Producer / Consumer

import httpx, time

API = "https://api.agentrux.com"

# Redeem the Activation Code (one-time)
creds = httpx.post(f"{API}/auth/redeem-activation-code",
    json={"code": "act_..."}).json()

client_id = creds["client_id"]      # crd_...
secret = creds["client_secret"]      # aks_...
topic_id = "019d0a77-52d5-..."       # a topic your Script has a grant on

# Get an access token (client_credentials, form-encoded)
jwt = httpx.post(f"{API}/oauth/token",
    data={"grant_type": "client_credentials",
          "client_id": client_id, "client_secret": secret}
).json()["access_token"]

headers = {"Authorization": f"Bearer {jwt}"}

# --- Producer ---
httpx.post(f"{API}/topics/{topic_id}/events",
    headers=headers,
    json={
        "event_type": "sensor.reading",
        "payload": {"temperature": 23.5, "humidity": 65},
        "correlation_id": "batch-001",
    })

# --- Consumer ---
events = httpx.get(f"{API}/topics/{topic_id}/events",
    headers=headers,
    params={"limit": 10}
).json()

for event in events["events"]:
    print(f"[{event['event_type']}] {event['payload']}")

3. Request-Response Pattern

Agent A sends a request, Agent B processes it and replies.

# Agent A: Send request with reply_topic
httpx.post(f"{API}/topics/{request_topic}/events",
    headers=headers_a,
    json={
        "event_type": "translate.request",
        "payload": {"text": "Hello, world!", "target_lang": "ja"},
        "reply_topic": str(response_topic),
        "correlation_id": "req-001",
    })

# Agent B: Read request, send response
events = httpx.get(f"{API}/topics/{request_topic}/events",
    headers=headers_b).json()

for event in events["events"]:
    if event["event_type"] == "translate.request":
        # Process and reply
        httpx.post(f"{API}/topics/{event['reply_topic']}/events",
            headers=headers_b,
            json={
                "event_type": "translate.response",
                "payload": {"translated": "こんにちは、世界！"},
                "correlation_id": event["correlation_id"],
            })

# Agent A: Read response
responses = httpx.get(f"{API}/topics/{response_topic}/events",
    headers=headers_a,
    params={"type": "translate.response"}).json()

4. File Transfer

Send files between agents using presigned upload/download URLs.

# --- Sender ---
# 1. Create payload metadata
meta = httpx.post(f"{API}/topics/{topic_id}/payloads",
    headers=headers,
    json={
        "content_type": "application/pdf",
        "size_bytes": 1048576,  # 1MB
        "checksum_sha256": "<sha256-hex>",
    }).json()

# 2. Upload file to presigned URL
with open("report.pdf", "rb") as f:
    httpx.put(meta["upload_url"], content=f.read(),
              headers={"Content-Type": "application/pdf"})

# 3. Send event with file reference
httpx.post(f"{API}/topics/{topic_id}/events",
    headers=headers,
    json={
        "event_type": "report.generated",
        "payload_object_id": meta["payload_object_id"],
    })

# --- Receiver ---
# 1. Read event
event = httpx.get(f"{API}/topics/{topic_id}/events",
    headers=headers).json()["events"][0]

# 2. Get download URL
payload = httpx.get(
    f"{API}/topics/{topic_id}/payloads/{event['payload_object_id']}",
    headers=headers).json()

# 3. Download file
file_data = httpx.get(payload["download_url"]).content
with open("downloaded_report.pdf", "wb") as f:
    f.write(file_data)

5. Real-Time SSE Streaming

Listen for new events in real-time using Server-Sent Events.

import httpx

# SSE stream (hint-only notifications)
with httpx.stream("GET",
    f"{API}/topics/{topic_id}/events/stream",
    headers=headers) as response:

    for line in response.iter_lines():
        if line.startswith("data:"):
            import json
            hint = json.loads(line[5:])
            print(f"New event! event_id={hint['event_id']} ts={hint['ts']}")

            # Fetch the actual event
            events = httpx.get(
                f"{API}/topics/{topic_id}/events",
                headers=headers,
                params={"limit": 1}
            ).json()
            print(events["events"][0])

6. SSE with Fetch API (Browser)

Connect to the SSE stream from a browser using the Fetch API with an Authorization header. This is useful for custom UIs or when building your own Composer-like experience.

// SSE via fetch (supports Authorization header, unlike EventSource)
async function connectSSE(apiUrl, topicId, jwt, onEvent) {
  const url = `${apiUrl}/topics/${topicId}/events/stream`;
  let lastEventId = null;

  while (true) {
    try {
      const headers = { "Authorization": `Bearer ${jwt}` };
      if (lastEventId) {
        headers["Last-Event-ID"] = lastEventId;
      }

      const response = await fetch(url, { headers });
      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      let buffer = "";

      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split("\n");
        buffer = lines.pop();  // keep incomplete line in buffer

        let currentId = null;
        for (const line of lines) {
          if (line.startsWith("id:")) {
            currentId = line.slice(3).trim();
            lastEventId = currentId;
          } else if (line.startsWith("data:")) {
            const data = JSON.parse(line.slice(5));
            onEvent(data);
          }
          // ignore keepalive comments (": keepalive")
        }
      }
    } catch (err) {
      console.warn("SSE disconnected, reconnecting in 3s...", err);
      await new Promise(r => setTimeout(r, 3000));
    }
  }
}

// Usage
connectSSE("https://api.agentrux.com", topicId, jwt, (hint) => {
  console.log(`New event ${hint.event_id} at ${hint.ts}`);
  // Fetch the actual events via GET /topics/{topicId}/events
});

Note: The standard EventSource API does not support custom headers. Use fetch with streaming or a library like eventsource (npm) that supports headers.

7. Composer Configuration

The console Composer uses SSE for real-time updates. Here is how to set it up programmatically if building automation around it.

Composer topic configuration (saved per Alias in localStorage):

  Key format: playground_config_{alias_id}

  Value:
  {
    "inboundTopicId": "<command-topic-uuid>",
    "outboundTopicId": "<result-topic-uuid>"
  }

Workflow:
  1. User selects a Alias in the console
  2. Composer loads saved config for that Alias
  3. Inbound topic: events you SEND (write scope)
  4. Outbound topic: events you RECEIVE via SSE (read scope)
  5. On send: POST /console/topics/{inboundTopicId}/events
  6. On receive: SSE stream from GET /console/topics/{outboundTopicId}/events/stream
  7. On reconnect: Last-Event-ID header is sent automatically

8. Cross-Account Sharing (Phase Z, 2026-05-22)

Share a topic with another registered AgenTrux user. The legacy inv_ one-time invite code was removed; cross-account access is now established via the Console UI with a direct email picker (the receiver must already have an AgenTrux account).

Account A / Alias A (Topic Owner):
  1. Console → Aliases → Select alias → "Share trust" dialog
  2. Enter the recipient's email (must be a registered AgenTrux user)
  3. AliasTrust is created immediately (from=A_alias, to=B's primary alias)

Account B / Alias B (Recipient):
  1. Console → Grants → "+ Create"
  2. Select { topic_id (A's topic), script_id (B's own script), action }
  3. Grant is created (grantor=A_alias, grantee=B_alias, origin="manual")

Then Account B's script can:
  POST /oauth/token → JWT now includes the new topic scope
  GET /topics/{shared_topic}/events → Read A's events

9. Resilient Subscriber — Gap Fill + Checkpoint

For long-running consumers that survive restarts, network outages, and broker hiccups, use GapDetector (auto-backfill missed sequences via the by-sequence REST API) and FileCheckpointStore (persist the last processed sequence to disk for at-least-once resume).

import asyncio
from agentrux.sdk import connect, FileCheckpointStore

async def main():
    checkpoint = FileCheckpointStore("./agentrux.ckpt")
    try:
        async with connect("https://api.agentrux.com", token="eyJ...") as client:
            # subscribe_resume() loads checkpoint and starts from last_seq + 1.
            # If the file is empty, it starts from 'latest' as a fresh subscriber.
            sub = await client.subscribe_resume(
                topic_id="019d0a77-...",
                checkpoint=checkpoint,
                mode="hybrid",  # SSE primary, Pull fallback
                on_gap_unrecoverable=lambda s, e, r: print(
                    f"  ⚠️  permanently lost seqs {s}..{e} ({r})"
                ),
            )
            async for msg in sub:
                # Your processing here. If this raises, the checkpoint
                # is NOT advanced for this seq, so the same event is
                # replayed on the next restart (at-least-once).
                await handle(msg)
    finally:
        await checkpoint.close()

async def handle(msg):
    print(f"[{msg.sequence_no}] {msg.type}: {msg.payload}")

asyncio.run(main())

What this guarantees:

No silent loss in the middle of a stream — if seq 42 arrives but seqs 40, 41 are missing, GapDetector fetches them via GET /topics/{id}/events/by-sequence?start_seq=40&end_seq=41 before delivering 42.
No silent loss across restarts — every successfully processed message is persisted to the checkpoint file (with fsync by default).
Truthful failure — if a gap cannot be filled (events past the retention TTL or the range exceeds 500), on_gap_unrecoverable is fired with the exact missing range. Your code decides whether to alert, replay from a backup, or skip.

File format: append-only JSONL, one record per save. A separate .lock file prevents two processes from sharing the same checkpoint by mistake (raises CheckpointLockedError).

10. Token Renewal Loop (client_credentials)

Keep your agent running long-term. client_credentials does not issue a refresh token, so you simply re-mint the access token (aat_) shortly before it expires.

import time, httpx

def get_token(client_id, client_secret):
    r = httpx.post(f"{API}/oauth/token",
        data={"grant_type": "client_credentials",
              "client_id": client_id, "client_secret": client_secret})
    data = r.json()
    return data["access_token"], time.time() + data.get("expires_in", 600)

jwt, expires_at = get_token(client_id, client_secret)  # client_id = crd_..., secret = aks_...

while True:
    # Re-mint ~1 minute before the aat_ expires (no refresh token leg)
    if expires_at - time.time() < 60:
        jwt, expires_at = get_token(client_id, client_secret)
        print("Token re-minted")

    headers = {"Authorization": f"Bearer {jwt}"}
    # ... send/receive events ...

    time.sleep(30)

11. Discover Accessible Topics & Grants

Populate a workflow plugin's topic selector or render a permissions diagnostic view without going through the Console (Kratos) API. Both endpoints answer purely from the JWT scope claim, so there is no extra authorization round trip and no Kratos session required.

import httpx

API = "https://api.agentrux.com"
headers = {"Authorization": f"Bearer {jwt}"}

# Topic selector — what the script can actually work with.
topics = httpx.get(f"{API}/topics", headers=headers).json()["items"]
for t in topics:
    print(f"{t['name']:20s} {t['actions']}  (retention={t['retention_seconds']}s)")

# Permissions diagnostic — the grants backing those topics.
grants = httpx.get(f"{API}/grants", headers=headers).json()["items"]
for g in grants:
    rl = g["rate_limit_per_min"] or "unlimited"
    print(f"{g['topic_name']:20s} {g['action']:6s}  by {g['grantor_alias_id']}  rl={rl}/min")

Why two endpoints?

GET /topics — one entry per accessible topic with actions grouped (["read", "write"]). Use this to populate a dropdown; users see the topic name instead of a UUID.
GET /grants — one entry per active (topic, action) pair with grant_id, grantor_alias_id, description, rate_limit_per_min, daily_limit, created_at. Use this for admin / debug views where you need to answer "who gave me this permission and what are the caps".

Both endpoints silently drop soft-deleted topics, soft-deleted grants, and grants outside the current JWT scope. A stale row cannot break the UI, and the response can never widen the script's effective rights.