Inference hooks統合の開発

Inference hooks統合とは、AIセキュリティサーバー、つまりAnthropicが呼び出すHTTPSサービスです。管理対象の各リクエストに対して、サーバーは会話のトランスクリプトを含む署名付きPOSTを受信し、許可または拒否の判定を返します。このページでは、そのサーバーを構築するためのプロトコル、すなわちリクエストと判定のスキーマ、署名検証、運用上の契約について説明します。

Inference hooksを有効にしてエンドポイントを指定する方法については、Inference hooksの設定を参照してください。Inference hooksの概要と使用すべき場面については、Inference hooksの概要を参照してください。



最初の判定ラウンドトリップを実行する

最小限の動作する統合は、各リクエストを読み取って許可するサーバーです。以下のいずれかのサーバーを実行し、パブリックなhttps:// URL（たとえば、TLS終端リバースプロキシまたはトンネルの背後）で公開してから、管理者にエンドポイントとして設定し、接続をテストしてもらいます。Test connection（接続テスト）の結果には、サーバーが返した許可判定が表示されます。

# 実行方法: python server.py
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


class VerdictHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"  # keep the connection open between verdicts

    def do_POST(self):
        # 本文を読み捨てます。トランスクリプトは数メガバイトになることがあります。
        self.rfile.read(int(self.headers.get("Content-Length", 0)))
        verdict = b'{"action": "allow"}'
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(verdict)))
        self.end_headers()
        self.wfile.write(verdict)


ThreadingHTTPServer(("", 8000), VerdictHandler).serve_forever()



リクエストを受信する

Anthropicは、管理者が設定したURLにHTTPS POSTを送信します。設定されたURL全体がエンドポイントです。固定のパスサフィックスはないため、サーバーに適した任意のパスを選択できます。

AIセキュリティサーバーは、Anthropicが到達できる場所にホストしてください。つまり、ポート443のhttps:// URLで、パブリックにルーティング可能なホスト上（プライベート、ループバック、キャリアグレードNATの範囲は接続時に拒否されます）にあり、パブリックCAトラストストアに対して検証される証明書を持ち、リダイレクトなしで応答する必要があります。設定されたURLは最終的な宛先でなければなりません。管理者がURLを設定およびテストする方法については、Inference hooksの設定を参照してください。

すべてのリクエストには、管理者が設定したカスタムリクエストヘッダーと、組織が署名シークレットを持っている場合は署名を検証するで説明されているwebhook-*署名ヘッダーに加えて、以下の固定ヘッダーが含まれます。

現在、フックイベントは1つだけです。プロンプトフレームは、管理対象の推論リクエストごとに1回、推論が開始される前に送信されます。Anthropicは、AIセキュリティサーバーが応答するか、判定タイムアウトが経過するまでリクエストを保留します。



プロンプトフレーム

リクエストボディは、以下のフィールドを持つJSONオブジェクトです。

リクエストボディの例：

{
  "type": "prompt",
  "request_id": "req_abc123",
  "tenant_id": "11111111-1111-1111-1111-111111111111",
  "actor": {
    "type": "user",
    "id": "user_01AbCdEfGhIjKlMnOpQrStUv",
    "email_address": "[email protected]"
  },
  "source": {
    "application": "claude-ai"
  },
  "session_id": "22222222-2222-2222-2222-222222222222",
  "model": "claude-sonnet-4-5",
  "messages": [
    {
      "role": "user",
      "content": [
        {
          "type": "text",
          "text": "Summarize the attached report."
        },
        {
          "type": "attachment",
          "file_name": "q2-report.pdf",
          "media_type": "application/pdf",
          "size_bytes": 48213,
          "text": "Q2 revenue grew 14% quarter over quarter..."
        }
      ]
    }
  ],
  "metadata": {}
}



コンテンツブロック

messagesの各エントリには、userまたはassistantのrole（ツール結果はuserロールの下に表示され、パブリックMessages APIのコンテンツモデルと一致します）と、typeで判別されるブロックのcontent配列があります。

認識できないtypeのブロックは、前方互換性のある追加です。保証される唯一のフィールドはtypeです。ポリシーは存在する他のフィールドを検査できますが、認識できないタイプを理由にリクエストを拒否してはなりません。



トランスクリプトに含まれる内容

トランスクリプトは、推論時点までのエンドユーザーが見る会話です。トランスクリプトテキスト、ツール呼び出しとその結果、抽出された添付ファイルテキスト、および以前のターンが含まれます。システムプロンプト、ツール定義、Anthropic内部コンテキスト、Claudeの隠された推論、生のファイルバイトは含まれません。

すべてのブロックが除外されたターンは完全に省略されるため、ユーザーとアシスタントが厳密に交互に現れることを前提としないでください。

トランスクリプトは切り詰められずに送信されるため、大きな添付ファイルを含む長い会話は、上限10 MBまでの大きなリクエストボディを生成します。その上限を受け入れるようにサーバーのボディ制限を引き上げてください。一般的なデフォルト値の多くはこれよりはるかに小さく、nginxのclient_max_body_sizeは1 MB、Expressのexpress.json()は100 kBです。拒否されたボディはWebhook失敗としてカウントされるため、Allow the request（リクエストを許可）の失敗処理では、サイズ超過のプロンプトが検査されずにモデルに到達することになります。



Sourceの値

source.applicationはオープンな文字列であり、閉じた列挙型ではありません。既知の値はclaude-aiとclaude-codeです。接続テストではconfig-testが使用されます。新しい値が出現する可能性があり、サーバーは認識できない値を理由にリクエストを拒否してはなりません。

source.applicationは、信頼境界ではなく、参考用のルーティングメタデータとして扱ってください。セキュリティ上重要なポリシー決定をこれだけに依存させないでください。



判定を返す

両方の結果に対して、HTTP 200とJSON判定ボディで応答します。actionフィールドで判別します。リクエストを許可する場合：

{
  "action": "allow"
}

拒否する場合：

{
  "action": "deny",
  "deny_reason": "This prompt appears to contain customer payment card data, which your organization's policy does not allow.",
  "reference_id": "scan_01HXPT4R9V"
}

拒否はフォーマットの問題で破棄されることはありません。サイズ超過のdeny_reasonは切り詰められ、不正な形式のreference_idは暗黙的に削除されますが、actionは引き続き尊重されます。

逆は成り立ちません。解析可能な判定を含むHTTP 200以外のものはすべてWebhook失敗であり、判定の代わりに組織の失敗処理が適用されます。特に：

• エラーステータスで拒否を通知しないでください。200以外のレスポンスは失敗であり、拒否ではありません。
• allowまたはdeny以外のaction値はWebhook失敗として扱われます。

Anthropicはレスポンスボディを最大64 KiBまで読み取り、ボディは非圧縮でなければなりません。リダイレクトは追跡されず、Cookieは無視されます。判定ボディ内の不明なフィールドは無視されるため、ここに記載されているフィールドと並んで、より豊富なオブジェクトを返すことができます。



署名を検証する

リクエストは、3つのヘッダーを使用してStandard Webhooks仕様に従って署名されます。Anthropicはヘッダー名を小文字で送信し、プロキシは大文字小文字を自由に変更できるため、大文字小文字を区別せずに検索してください。

検証バグのほとんどは、次の2つの詳細が原因です。

• 生のバイトを検証してください。 JSON解析や再エンコードの前に、受信したとおりのボディに対してHMACを計算します。
• 標準のbase64デコーダーでシークレットをデコードしてください。 署名シークレットはwhsec_プレフィックスの後の値で、標準のbase64アルファベット（+と/）でエンコードされています。ヘッダー内の署名も同様です。URLセーフデコーダーは、シークレットに+または/が含まれる場合（ほとんどの場合そうです）、誤ったキーバイトを導出します。

組織が署名シークレットを持つと、Anthropicが送信するすべてのリクエストは署名され、Inference hooksを有効にするには署名シークレットが必要なため、署名されずに到着したリクエストは拒否してください。1つの例外として、組織の最初の保存前に送信された接続テストは、署名シークレットがまだ存在しないため、署名されずに到着します。管理者がシークレットの存在を確認するまでは署名されていないリクエストを受け入れ、その後は拒否してください。

シークレットのローテーションは即座の切り替えですが、以前のシークレットで署名されたリクエストは、その後約1分間、および既に処理中のものが引き続き到着する可能性があります。切り替え中は、AIセキュリティサーバーが両方のシークレットからの署名を受け入れるようにして、これらの遅延リクエストが拒否されないようにしてください。

以下のサンプルはサーバー実装であるため、シェルタブはありません。AIセキュリティサーバーは、単発のリクエストではなく、長時間実行されるHTTPSサービスです。各サンプルは言語の標準ライブラリのみを使用しています。Standard Webhooksプロジェクトも、ほとんどの言語向けの検証ライブラリを公開しています。

import base64
import hashlib
import hmac
import time

TOLERANCE_SECONDS = 300


def verify(secret: str, headers: dict[str, str], body: bytes) -> bool:
    """Return True if the body was signed by Anthropic for this organization.

    Anthropic sends header names in lowercase, but proxies are free to
    re-case them, so normalize the lookup to lowercase.
    """
    lowercased = {name.lower(): value for name, value in headers.items()}
    try:
        message_id = lowercased["webhook-id"]
        timestamp = lowercased["webhook-timestamp"]
        signatures = lowercased["webhook-signature"]
    except KeyError:
        return False  # unsigned request: not from Anthropic

    try:
        signed_at = int(timestamp)
    except ValueError:
        return False
    if abs(time.time() - signed_at) > TOLERANCE_SECONDS:
        return False  # replayed, or the clocks disagree

    try:
        key = base64.b64decode(secret.removeprefix("whsec_"), validate=True)
    except ValueError:
        return False  # misconfigured secret: reject rather than crash

    payload = f"{message_id}.{timestamp}.".encode() + body
    expected = b"v1," + base64.b64encode(
        hmac.new(key, payload, hashlib.sha256).digest()
    )

    # バイト列を比較：strに対するcompare_digestは非ASCII入力でエラーを発生させます。
    return any(
        hmac.compare_digest(expected, candidate.encode())
        for candidate in signatures.split()
    )



運用上のセマンティクス



タイムアウトと再試行

管理者は、1〜10,000ms（デフォルトは5,000ms）の判定タイムアウトを設定します。この予算は、接続、TLSハンドシェイク、リクエスト、レスポンスを含む交換全体をカバーします。

Anthropicは、接続試行が失敗した場合にのみ、100msの遅延後に正確に1回再試行します。再試行は同じタイムアウト予算を共有し、同じwebhook-idと同じ署名を持ちます。AIセキュリティサーバーが応答した後は、交換は再試行されません。



Webhook失敗

タイムアウト、200以外のステータス（リダイレクトを含む）、解析不能またはサイズ超過のレスポンスボディ、到達不能なエンドポイントはすべてWebhook失敗です。Webhook失敗が拒否になることはありません。代わりに、組織の失敗処理設定により、影響を受けるリクエストがブロックされるか、検査なしで続行されるかが決まります。



サーキットブレーカー

AIセキュリティサーバーに起因する持続的なWebhook失敗は、適用を停止するサーキットブレーカーを作動させます。Anthropicはサーバーへの接続を停止し、すべてのリクエストに失敗処理が適用されます。復旧は管理者側で行います。サーバーを修正してから、管理者にEnforce verdicts（判定を適用）を再度オンにしてもらいます。サーキットブレーカーを参照してください。



レイテンシ

適用により、組織内のすべての管理対象リクエストの「latency」（レイテンシ）にAIセキュリティサーバーのラウンドトリップが追加されます。判定を高速に保ち、大規模な組織に展開する前にサーバーの負荷テストを行ってください。



送信元IPアドレス

AIセキュリティサーバーへのリクエストは、Anthropicの公開されているアウトバウンドIP範囲の一部である160.79.106.0/24から発信されます。同じページのインバウンド範囲ではなく、このブロックを許可リストに追加してください。インバウンド範囲はこれをカバーしていません。許可リストはサーバーの露出を狭めますが、署名検証の代替にはなりません。このブロックはInference hooks以外のAnthropicのエグレストラフィックも運びます。



前方互換性

プロトコルは、正しく記述されたサーバーを壊すことなく拡張されます。サーバーは以下を無視する必要があります。

• プロンプトフレームの不明なトップレベルフィールド。
• metadata内の不明なキー。
• 新しいsource.application値。
• 新しいactor.type値。actorはtypeで判別されるユニオンであり、現在送信される種類は"user"のみです。将来の種類はtypeが存在することのみを保証します。
• 認識できないtypeのコンテンツブロック。

認識できないブロックタイプやフィールドを理由にリクエストを拒否しないでください。知っているフィールドを読み取り、残りはスキップしてください。

将来的に他のフックイベントタイプが導入されます。新しいイベントタイプは、フィールドをスキップすることでは処理できない追加です。リクエストには依然として判定が必要です。トップレベルのtypeが認識できない値の場合は、エラーステータスではなく許可判定を返してください。エラーレスポンスはWebhook失敗であり、持続的な失敗はサーキットブレーカーを作動させます。



統合を設計する

本番環境のAIセキュリティサーバーは、ワイヤープロトコル以外にもいくつかの設計上の選択を行います。

webhook-idで重複排除してください。 webhook-idヘッダーは配信ごとに一意であり、ボディのrequest_idと同じ値です。接続失敗の再試行はこれを再利用するため、冪等性キーとして機能します。判定を記録する場合は、これをキーとしてレコードを作成してください。

判定を記録し、拒否を結合してください。 返した各判定をreference_idとともに保存します。すべての拒否は、サーバーが返したreference_idを含むinference_hooks_request_deniedコンプライアンスアクティビティとして記録されるため、アクティビティフィードの拒否を自分のシステム内の対応するレコードと結合できます。

常に許可するサーバーでアーカイブしてください。 トランスクリプトを監視せずにリアルタイムでキャプチャするには、無条件に{"action": "allow"}を返し、応答後にフレームを永続化します。これはCompliance APIをポーリングする代わりのプッシュベースの方法であり、永続化する前に応答することで、ラウンドトリップをユーザーのクリティカルパスから外すことができます。

エンドユーザー向けにdeny_reasonを記述してください。 返すテキストは、リクエストがブロックされたときにユーザーが見るもので、500文字で切り詰められます。チームだけが解釈できるスキャナーコードを出力するのではなく、削除すべきコンテンツの種類など、何を変更すべきかを伝えてください。



次のステップ