Skip to main content

Anthropic Integration

Route your Anthropic Claude API traffic through Rivaro for runtime enforcement. SDK configuration, the Messages endpoint, streaming, and tool use.

SDK Configuration​

Python​

from anthropic import Anthropic

client = Anthropic(
    api_key="sk-ant-your-key",
    base_url="https://your-org.rivaro.ai",
    default_headers={
        "X-Detection-Key": "detect_live_your_key_here"
    }
)
note

The base_url does not include /v1 — the Anthropic SDK adds the /v1/messages path automatically.

Node.js​

import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic({
  apiKey: 'sk-ant-your-key',
  baseURL: 'https://your-org.rivaro.ai',
  defaultHeaders: {
    'X-Detection-Key': 'detect_live_your_key_here'
  }
});

Supported Endpoints​

EndpointMethodDescription
/v1/messagesPOSTCreate a message (Claude 3, Claude 3.5, Claude 4)

Request and response formats match the Anthropic API exactly.

Required Headers​

HeaderRequiredDescription
X-Detection-KeyYesYour Rivaro detection key
x-api-keyYesYour Anthropic API key
anthropic-versionYesAnthropic API version (e.g. 2023-06-01)
Content-TypeYesapplication/json

The anthropic-version header is always included by Rivaro (defaults to 2023-06-01). If you send a different version in your request, yours takes precedence.

Streaming​

Streaming works out of the box. Rivaro forwards text delta events to your application in real time.

with client.messages.stream(
    model="claude-sonnet-4-5-20250929",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain quantum computing"}]
) as stream:
    for text in stream.text_stream:
        print(text, end="")

How enforcement interacts with streaming​

  • Text deltas (content_block_delta events with delta.type: "text_delta") are forwarded to your application immediately.
  • Egress detection runs on the accumulated full response after message_stop.
  • Ingress detection on your messages runs before the request is forwarded to Anthropic. If a policy blocks the input, Anthropic is never called.

Anthropic SSE format​

Anthropic uses a slightly different SSE format than OpenAI:

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}

event: message_stop
data: {"type":"message_stop"}

Rivaro handles this format transparently — you don't need to do anything different.

Tool Use​

Anthropic's tool use works through the proxy. Rivaro inspects tool definitions in your request and tool use blocks in the response.

response = client.messages.create(
    model="claude-sonnet-4-5-20250929",
    max_tokens=1024,
    tools=[{
        "name": "get_weather",
        "description": "Get current weather for a city",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string"}
            },
            "required": ["city"]
        }
    }],
    messages=[{"role": "user", "content": "What's the weather in London?"}]
)

What Rivaro does with tool use​

  • Request side: Rivaro reads the tools[] array. If a policy blocks specific tools, those are filtered before the request reaches Anthropic.
  • Response side: Rivaro inspects content[] blocks where type == "tool_use". Each tool's name and input are checked against detection rules.
  • Streaming: During streaming, content_block_start events with type: "tool_use" are accumulated. Detection runs after the stream completes.

Blocked Requests​

When Rivaro blocks a request, the response matches Anthropic's format:

{
  "content": [{"type": "text", "text": "Content blocked due to policy violations"}],
  "stop_reason": "content_filtered"
}

Check for stop_reason: "content_filtered" to detect enforcement blocks.

Next steps​