Skip to content

Anthropic SDK Integration

Swap AsyncAnthropic for MendGuardrailsAsyncAnthropic (or Anthropic for MendGuardrailsAnthropic). Everything else stays the same — the guardrail pipeline (pre-flight → input → LLM → output) runs automatically around every messages.create call.

Install the optional dependency:

pip install 'mend-guardrails[anthropic]'

import asyncio
from mendguardrails import MendGuardrailsAsyncAnthropic, GuardrailEnforcementTriggered

async def main():
    client = MendGuardrailsAsyncAnthropic(name="Client 1")

    try:
        response = await client.messages.create(
            model="claude-3-5-sonnet-latest",
            max_tokens=512,
            messages=[{"role": "user", "content": "Hello, can you help me?"}],
        )
        text = "".join(block.text for block in response.content if block.type == "text")
        print(text)

    except GuardrailEnforcementTriggered as exc:
        print(f"Blocked by: {exc.guardrail_result.info['guardrail_name']}")

asyncio.run(main())

Multi-turn conversations

Pass the current user message inline — don't append to history until after guardrails pass. This prevents blocked messages from persisting in your conversation context.

messages: list[dict] = []

while True:
    user_input = input("You: ")

    try:
        # Pass user message inline, not pre-appended to messages
        response = await client.messages.create(
            model="claude-3-5-sonnet-latest",
            max_tokens=512,
            messages=messages + [{"role": "user", "content": user_input}],
            suppress_enforcement=False,
        )

        content = "".join(b.text for b in response.content if b.type == "text")
        print(f"Assistant: {content}")

        # Only append after guardrails pass
        messages.append({"role": "user", "content": user_input})
        messages.append({"role": "assistant", "content": content})

    except GuardrailEnforcementTriggered:
        print("Message blocked — not added to history")

Streaming

Set stream=True to receive Anthropic streaming events. Output guardrails run on the accumulated response; if a block policy fires mid-stream a GuardrailEnforcementTriggered is raised.

stream = await client.messages.create(
    model="claude-3-5-sonnet-latest",
    max_tokens=512,
    messages=[{"role": "user", "content": "Write a short poem."}],
    stream=True,
)

async for event in stream:
    if event.type == "content_block_delta":
        print(event.delta.text, end="", flush=True)

Third-party / self-hosted Claude endpoints

MendGuardrailsAsyncAnthropic accepts the same constructor arguments as AsyncAnthropic, so you can point it at any Anthropic-compatible endpoint (for example Amazon Bedrock or Google Vertex via a compatible gateway) with base_url and api_key:

client = MendGuardrailsAsyncAnthropic(
    base_url="https://your-anthropic-gateway.example.com",
    api_key="your-key",
)

Next steps