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
| Parameter | Type | Description |
|---|---|---|
messages | list[Message \| dict] | Conversation history as Message objects or dicts |
guardrails | list[GuardrailConfigType] | List of guardrail policies to apply |
target | GuardrailsTarget | What 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:
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
| Role | Description |
|---|---|
Role.SYSTEM | System instructions |
Role.USER | User messages |
Role.ASSISTANT | LLM responses |
Role.TOOL | Tool/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
| Exception | When it's raised | Recommended action |
|---|---|---|
GuardrailsTriggered | A guardrail detected a violation | Block the request, log the violation |
GuardrailsAPITimeoutError | Request exceeded timeout | Retry or implement fail-open/fail-closed |
GuardrailsAPIConnectionError | Network connectivity issues | Retry with backoff, alert on-call |
GuardrailsAPIResponseError | API returned non-2xx status | Log error, check API status |
GuardrailsConnectionTestError | test_connection() failed | Check credentials and endpoint configuration |
To disable exception throwing on guardrail trigger (fail-open behavior):