AgenTrux — 使い方の例

エージェント間通信の実践的なパターン集です。

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

1. 基本: redeem → 送信 → 受信

最もシンプルなフロー。1つのエージェントが送信し、もう1つが受信します。

ステップ 1: Activation Code を redeem

curl -X POST https://api.agentrux.com/auth/redeem-activation-code \
  -H "Content-Type: application/json" \
  -d '{"code": "act_あなたのアクティベーションコード"}'

レスポンス:

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

client_id（crd_）と client_secret（aks_）は永続的に保存してください。aks_ は 1 度だけ表示され、同じコードの 2 回目の redeem はできません。

ステップ 2: アクセストークンを取得

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'

ステップ 3: イベントを送信

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": "Q1の売上データを分析して",
      "priority": "high",
      "deadline": "2026-03-25T00:00:00Z"
    }
  }'

ステップ 4: イベントを受信

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

2. 常駐型エージェントへのコマンド投入

管理者がコマンドを送信し、常駐エージェントが処理して結果を返すパターン。

"""
構成:
  command_topic  ← 管理者がコマンドを投入
  result_topic   ← エージェントが結果を返す

管理者 → command_topic → 常駐エージェント → result_topic → 管理者
"""
import asyncio, json, httpx, time
from datetime import datetime, timezone

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

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# 常駐エージェント（バックグラウンドで動き続ける）
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

async def resident_agent(client_id, client_secret, command_topic, result_topic):
    """コマンドを待ち受け、処理して結果を返す常駐エージェント"""

    # アクセストークン取得（client_credentials、form-encoded）
    r = httpx.post(f"{API}/oauth/token",
                   data={"grant_type": "client_credentials",
                         "client_id": client_id, "client_secret": client_secret})
    jwt = r.json()["access_token"]
    expires_at = time.time() + r.json().get("expires_in", 600)
    headers = {"Authorization": f"Bearer {jwt}"}

    last_cursor = None
    print("🤖 エージェント起動。コマンド待機中...")

    while True:
        # client_credentials は refresh token なし。期限近くで再取得（約1分前）
        if expires_at - time.time() < 60:
            r = httpx.post(f"{API}/oauth/token",
                           data={"grant_type": "client_credentials",
                                 "client_id": client_id, "client_secret": client_secret})
            jwt = r.json()["access_token"]
            expires_at = time.time() + r.json().get("expires_in", 600)
            headers = {"Authorization": f"Bearer {jwt}"}
            print("🔄 トークンリフレッシュ完了")

        # コマンドをポーリング
        params = {"limit": 10, "type": "command"}
        if last_cursor:
            params["after"] = last_cursor

        r = httpx.get(f"{API}/topics/{command_topic}/events",
                      headers=headers, params=params)
        events = r.json()

        for event in events["events"]:
            cmd = event["payload"]
            correlation = event.get("correlation_id", "")
            print(f"📥 コマンド受信: {cmd.get('action')} (correlation={correlation})")

            # コマンドを処理
            result = process_command(cmd)

            # 結果を返す
            httpx.post(f"{API}/topics/{result_topic}/events",
                       headers=headers,
                       json={
                           "event_type": "command.result",
                           "payload": result,
                           "correlation_id": correlation,
                       })
            print(f"📤 結果送信完了: {result.get('status')}")

        next_cursor = events.get("next", {}).get("after")
        if next_cursor:
            last_cursor = next_cursor

        await asyncio.sleep(2)  # 2秒間隔でポーリング

def process_command(cmd):
    """コマンドを処理して結果を返す"""
    action = cmd.get("action")

    if action == "analyze":
        # 分析処理（実際にはAI呼び出し等）
        return {
            "status": "success",
            "action": action,
            "result": f"「{cmd.get('target', '')}」の分析完了。売上は前年比120%。",
            "completed_at": datetime.now(timezone.utc).isoformat(),
        }
    elif action == "report":
        return {
            "status": "success",
            "action": action,
            "result": "レポートを生成しました。",
            "report_url": "https://example.com/report.pdf",
        }
    elif action == "status":
        return {
            "status": "success",
            "action": action,
            "uptime": "3h 42m",
            "processed_commands": 156,
            "last_error": None,
        }
    else:
        return {"status": "error", "message": f"Unknown action: {action}"}

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# 管理者（コマンドを投入する側）
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

def send_command(headers, command_topic, action, **kwargs):
    """管理者がエージェントにコマンドを送信"""
    import uuid
    correlation_id = f"cmd-{uuid.uuid4().hex[:8]}"

    r = httpx.post(f"{API}/topics/{command_topic}/events",
                   headers=headers,
                   json={
                       "event_type": "command",
                       "payload": {"action": action, **kwargs},
                       "correlation_id": correlation_id,
                   })
    print(f"📤 コマンド送信: {action} (correlation={correlation_id})")
    return correlation_id

def wait_for_result(headers, result_topic, correlation_id, timeout=30):
    """結果を待つ"""
    import time
    start = time.time()
    while time.time() - start < timeout:
        r = httpx.get(f"{API}/topics/{result_topic}/events",
                      headers=headers, params={"type": "command.result"})
        for event in r.json()["events"]:
            if event.get("correlation_id") == correlation_id:
                return event["payload"]
        time.sleep(1)
    return {"status": "timeout"}

# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# 使用例
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

# 管理者がコマンドを投入:
#   cid = send_command(headers, command_topic, "analyze", target="Q1売上データ")
#   result = wait_for_result(headers, result_topic, cid)
#   print(result)
#   → {"status": "success", "result": "「Q1売上データ」の分析完了。売上は前年比120%。"}

# ステータス確認:
#   cid = send_command(headers, command_topic, "status")
#   result = wait_for_result(headers, result_topic, cid)
#   print(result)
#   → {"status": "success", "uptime": "3h 42m", "processed_commands": 156}

3. リクエスト-レスポンスパターン

エージェント A がリクエストを送り、エージェント B が処理して返答。

# エージェント A: 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",
    })

# エージェント B: リクエストを読み、レスポンスを返す
events = httpx.get(f"{API}/topics/{request_topic}/events",
    headers=headers_b).json()

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

4. ファイル転送

署名付き URL を使ってエージェント間でファイルを送受信。

# --- 送信側 ---
# 1. ペイロードメタデータを作成
meta = httpx.post(f"{API}/topics/{topic_id}/payloads",
    headers=headers,
    json={"content_type": "application/pdf", "size_bytes": 1048576,
          "checksum_sha256": "<sha256-hex>"}).json()

# 2. 署名付き URL にファイルをアップロード
with open("report.pdf", "rb") as f:
    httpx.put(meta["upload_url"], content=f.read(),
              headers={"Content-Type": "application/pdf"})

# 3. ファイル参照付きイベントを送信
httpx.post(f"{API}/topics/{topic_id}/events",
    headers=headers,
    json={"event_type": "report.generated", "payload_object_id": meta["payload_object_id"]})

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

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

file_data = httpx.get(payload["download_url"]).content

5. SSE リアルタイムストリーミング

Server-Sent Events で新しいイベントをリアルタイムに検知。

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:"):
            hint = json.loads(line[5:])
            print(f"新イベント! event_id={hint['event_id']} ts={hint['ts']}")

            # 実際のイベントを取得
            events = httpx.get(f"{API}/topics/{topic_id}/events",
                headers=headers, params={"limit": 1}).json()
            print(events["events"][0])

6. Fetch API による SSE 接続（ブラウザ）

ブラウザの Fetch API を使って Authorization ヘッダ付きで SSE ストリームに接続します。独自の UI や Composer 風の体験を構築する際に有用です。

// fetch による SSE（EventSource と異なり Authorization ヘッダを送れる）
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();  // 不完全な行はバッファに保持

        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);
          }
          // keepalive コメント (": keepalive") は無視
        }
      }
    } catch (err) {
      console.warn("SSE 切断、3秒後に再接続...", err);
      await new Promise(r => setTimeout(r, 3000));
    }
  }
}

// 使用例
connectSSE("https://api.agentrux.com", topicId, jwt, (hint) => {
  console.log(`新イベント: event_id=${hint.event_id} ts=${hint.ts}`);
  // GET /topics/{topicId}/events で実際のイベントを取得
});

補足: 標準の EventSource API はカスタムヘッダに対応していません。fetch のストリーミングか、ヘッダをサポートする eventsource（npm）などのライブラリを使用してください。

7. Composer の設定

コンソールの Composer は SSE を使ったリアルタイム更新に対応しています。自動化を構築する場合の設定方法を以下に示します。

Composer トピック設定（Alias ごとに localStorage に保存）:

  キー形式: playground_config_{alias_id}

  値:
  {
    "inboundTopicId": "<コマンドトピックの UUID>",
    "outboundTopicId": "<結果トピックの UUID>"
  }

フロー:
  1. コンソールで Alias を選択
  2. Composer がその Alias の保存済み設定を読み込む
  3. Inbound topic: 送信するイベント（write スコープ）
  4. Outbound topic: SSE で受信するイベント（read スコープ）
  5. 送信時: POST /console/topics/{inboundTopicId}/events
  6. 受信時: GET /console/topics/{outboundTopicId}/events/stream の SSE ストリーム
  7. 再接続時: Last-Event-ID ヘッダが自動送信される

8. クロスアカウント共有（Phase Z、 2026-05-22）

登録済の別 AgenTrux user に Topic アクセスを共有します。 旧 inv_ 招待コードは廃止され、 Console UI で直接 email 指定する 2 段操作に統一されました（受領者は事前に AgenTrux 登録必須）。

アカウント A / Alias A（トピック所有者）:
  1. コンソール → Aliases → 対象 alias 選択 → 「Share trust」 ダイアログ
  2. 受領者の email を入力（登録済 AgenTrux user 必須）
  3. AliasTrust 行が即作成される (from=A_alias, to=B の primary alias)

アカウント B / Alias B（受け取る側）:
  1. コンソール → Grants → 「+ Create」
  2. { topic_id (A の topic), script_id (B 自身の script), action } を指定
  3. Grant 行が作成される (grantor=A_alias, grantee=B_alias, origin="manual")

その後、 アカウント B の script は:
  POST /oauth/token → JWT に新しい topic scope が含まれる
  GET /topics/{共有 topic}/events で A のイベントを読み取り可能

9. 断線復旧パターン — Gap Fill + Checkpoint

長時間稼働するコンシューマで再起動・ネットワーク断・ブローカー障害を 跨いで at-least-once を保証するには、GapDetector（by-sequence REST API で欠損シーケンスを自動補填）と FileCheckpointStore（最後に 処理した sequence_no をディスクに永続化）を組み合わせます。

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() は checkpoint を load して last_seq+1 から再開する。
            # 初回起動時 (checkpoint が空) は 'latest' から購読する。
            sub = await client.subscribe_resume(
                topic_id="019d0a77-...",
                checkpoint=checkpoint,
                mode="hybrid",  # SSE 優先、Pull フォールバック
                on_gap_unrecoverable=lambda s, e, r: print(
                    f"  ⚠️  復旧不能な欠損 seq {s}..{e} ({r})"
                ),
            )
            async for msg in sub:
                # ユーザ処理。例外を投げると checkpoint は前進せず、
                # 再起動後に同じイベントが再配信される（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())

この実装が保証すること:

• ストリーム途中の欠損を補填 — seq 42 が届いた時に 40, 41 が抜けて いれば、GapDetector が GET /topics/{id}/events/by-sequence?start_seq=40&end_seq=41 で取得してから 42 を配信する。
• 再起動を跨いだ欠損なし — 正常処理されたメッセージは毎回 checkpoint ファイルに fsync 付きで保存される。
• 正直な失敗通知 — retention TTL を越えた / range が 500 件超など の理由で取り戻せない場合は on_gap_unrecoverable が正確な範囲付きで 呼ばれる。アプリ側でアラート・バックアップ参照・スキップなどを判断する。

ファイル形式: 1レコード1行の追記専用 JSONL。専用の .lock ファイル で多重起動を防止（2番目以降のプロセスは CheckpointLockedError）。

10. トークン再取得ループ（client_credentials）

長時間稼働するエージェント向け。client_credentials は refresh token を発行しないため、aat_ の期限が近づいたら再取得します。

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

jwt, expires_at = get_token()
while True:
    # 期限約1分前に再取得（refresh token なし）
    if expires_at - time.time() < 60:
        jwt, expires_at = get_token()

    # 通常の処理
    headers = {"Authorization": f"Bearer {jwt}"}
    # ... イベント送受信 ...
    time.sleep(30)

11. Topic / Grant のディスカバリ

ワークフロープラグインのトピックセレクタ表示や、権限の診断ビューを Console(Kratos)API を経由せずに取得します。どちらのエンドポイントも JWT の scope クレームだけで応答するので、追加の認可ラウンドトリップ も Kratos セッションも不要です。

import httpx

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

# トピックセレクタ — スクリプトが実際に使えるトピック一覧
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)")

# 権限の診断 — それぞれの topic を裏付ける grant の一覧
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")

なぜ 2 つのエンドポイントか?

• GET /topics — アクセス可能なトピック 1 件につき 1 エントリ、actions は ["read", "write"] の形でまとめられます。ドロップダウン表示用で、 UUID ではなくトピック名を見せられます。
• GET /grants — アクティブな (topic, action) の組み合わせ 1 件につき 1 エントリ、grant_id / grantor_alias_id / description / rate_limit_per_min / daily_limit / created_at 付き。「この権限を 誰に・どの条件で・いつ付与されたか」を表示する管理画面や診断 UI 用。

どちらもソフト削除済みのトピック / grant、および JWT scope 範囲外の grant は静かに除外されます。古いレコードが UI を壊すことはなく、かつ このエンドポイント経由でスクリプトの実効権限が広がることもありません。