Skip to content

Guard API

The guard() method provides full control over message history, making it ideal for multi-turn conversations and complex guardrail scenarios.

Methods overview

guard_prompt()

Guards user input before sending to the LLM. See Getting Started with Guardrails for examples.

guard_response()

Guards LLM output after generation. Optionally accepts the original prompt for context-aware analysis. See Getting Started with Guardrails for examples.

The guard() method

The guard() API accepts a list of messages with conversation context:

await guardrails.guard(
    messages=messages,
    guardrails=config,
    target=GuardrailsTarget.PROMPT,  # or GuardrailsTarget.RESPONSE
)

Parameters

ParameterTypeDescription
messageslist[Message \| dict]Conversation history as Message objects or dicts
guardrailslist[GuardrailConfigType]List of guardrail policies to apply
targetGuardrailsTargetWhat to guard: PROMPT or RESPONSE

Target types

  • GuardrailsTarget.PROMPT — Guards the latest user message in the conversation.
  • GuardrailsTarget.RESPONSE — Guards the latest assistant message (must be the last message).

Basic usage

Guard a user prompt with conversation context:

import asyncio
from cx_guardrails import Guardrails, PII, PromptInjection, GuardrailsTarget, GuardrailsTriggered

async def main():
    guardrails = Guardrails()

    messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is machine learning?"},
        {"role": "assistant", "content": "Machine learning is a subset of AI..."},
        {"role": "user", "content": "Can you explain neural networks?"},
    ]

    async with guardrails.guarded_session():
        try:
            await guardrails.guard(
                messages=messages,
                guardrails=[PII(), PromptInjection()],
                target=GuardrailsTarget.PROMPT,
            )
            print("✓ Prompt passed")
        except GuardrailsTriggered as e:
            print(f"✗ Blocked: {e}")

asyncio.run(main())

Expected output:

✓ Prompt passed

Conversations with tool calls

When your LLM uses tools/functions, include tool messages in the conversation history. Tool call details and tool results should be in the content field:

import asyncio
from cx_guardrails import Guardrails, PII, PromptInjection, GuardrailsTarget, GuardrailsTriggered

async def main():
    guardrails = Guardrails()

    messages = [
        {"role": "system", "content": "You are a helpful assistant with access to tools."},
        {"role": "user", "content": "What's the weather in San Francisco?"},
        {"role": "assistant", "content": '[tool_call: get_weather({"location": "San Francisco"})]'},
        {"role": "tool", "content": '{"temperature": 65, "condition": "sunny"}'},
        {"role": "assistant", "content": "The weather in San Francisco is sunny with a temperature of 65°F."},
        {"role": "user", "content": "Thanks! Now what about New York?"},
    ]

    async with guardrails.guarded_session():
        try:
            await guardrails.guard(
                messages=messages,
                guardrails=[PII(), PromptInjection()],
                target=GuardrailsTarget.PROMPT,
            )
            print("✓ Prompt passed")
        except GuardrailsTriggered as e:
            print(f"✗ Blocked: {e}")

asyncio.run(main())

Building tool call conversations incrementally

messages = [
    {"role": "user", "content": "What's the weather in Paris?"},
]

messages.append({"role": "assistant", "content": '[tool_call: get_weather({"location": "Paris"})]'})
messages.append({"role": "tool", "content": '{"temperature": 18, "condition": "cloudy"}'})
messages.append({"role": "assistant", "content": "The weather in Paris is cloudy with a temperature of 18°C."})

Using the Message class

For type safety and better IDE support, use the Message class instead of plain dicts:

from cx_guardrails import Guardrails, Message, Role, PII, GuardrailsTarget

async def main():
    guardrails = Guardrails()

    messages = [
        Message(role=Role.SYSTEM, content="You are a helpful assistant with access to tools."),
        Message(role=Role.USER, content="What's the weather in San Francisco?"),
        Message(role=Role.ASSISTANT, content='[tool_call: get_weather({"location": "San Francisco"})]'),
        Message(role=Role.TOOL, content='{"temperature": 65, "condition": "sunny"}'),
        Message(role=Role.ASSISTANT, content="The weather in San Francisco is sunny with a temperature of 65°F."),
    ]

    async with guardrails.guarded_session():
        await guardrails.guard(
            messages=messages,
            guardrails=[PII()],
            target=GuardrailsTarget.PROMPT,
        )

Available roles

RoleDescription
Role.SYSTEMSystem instructions
Role.USERUser messages
Role.ASSISTANTLLM responses
Role.TOOLTool/function call results

Full guarded conversation example

import asyncio
from openai import AsyncOpenAI
from cx_guardrails import Guardrails, Message, Role, PII, PromptInjection, GuardrailsTarget, GuardrailsTriggered

async def main():
    guardrails = Guardrails()
    openai_client = AsyncOpenAI()

    messages = [
        Message(role=Role.SYSTEM, content="You are a helpful assistant."),
        Message(role=Role.USER, content="What is AI observability? Explain in one sentence."),
    ]

    async with guardrails.guarded_session():
        try:
            await guardrails.guard(
                messages=messages,
                guardrails=[PII(), PromptInjection()],
                target=GuardrailsTarget.PROMPT,
            )
            print("✓ User input passed")
        except GuardrailsTriggered as e:
            return print(f"✗ Blocked: {e}")

        openai_messages = [{"role": m.role.value, "content": m.content} for m in messages]
        response = await openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=openai_messages,
        )
        llm_response = response.choices[0].message.content

        messages.append(Message(role=Role.ASSISTANT, content=llm_response))

        try:
            await guardrails.guard(
                messages=messages,
                guardrails=[PII()],
                target=GuardrailsTarget.RESPONSE,
            )
            print("✓ LLM response passed")
        except GuardrailsTriggered as e:
            return print(f"✗ Response blocked: {e}")

        print(f"\n📝 AI RESPONSE:\n{llm_response}")

asyncio.run(main())

Expected output:

✓ User input passed
✓ LLM response passed

📝 AI RESPONSE:
AI observability refers to the tools and practices used to monitor, analyze, and understand the behavior and performance of AI models and systems in real-time.

Error types

ExceptionWhen it's raisedRecommended action
GuardrailsTriggeredA guardrail detected a violationBlock the request, log the violation
GuardrailsAPITimeoutErrorRequest exceeded timeoutRetry or implement fail-open/fail-closed
GuardrailsAPIConnectionErrorNetwork connectivity issuesRetry with backoff, alert on-call
GuardrailsAPIResponseErrorAPI returned non-2xx statusLog error, check API status
GuardrailsConnectionTestErrortest_connection() failedCheck credentials and endpoint configuration

To disable exception throwing on guardrail trigger (fail-open behavior):

export DISABLE_GUARDRAILS_TRIGGERED_EXCEPTION=true