Custom Guardrails
Write synchronous or asynchronous Python checks for validation, authorization, and application-specific safety rules.
Create a guardrail
Write Python files in a guardrails/ directory, following the same project pattern as middleware. Custom checks use the modes and execution order described in the guardrails guide.
from connic import GuardrailResult
import re
def check(content: str, context: dict) -> GuardrailResult:
"""Verify the input contains a valid ticket ID format."""
if not re.search(r'TICKET-\d{4,8}', content):
return GuardrailResult(
passed=False,
message="Please include a valid ticket ID (e.g., TICKET-12345)"
)
return GuardrailResult(passed=True)Contract
- File name matches the
namein config (e.g.,guardrails/validate-ticket-id.pyforname: validate-ticket-id) - Must export a
check(content: str, context: dict) -> GuardrailResultfunction (sync or async) contentis the text being checked (input or output)contextis the same context dict available to middleware (run_id,agent_name,connector_id,timestamp, and any user-set values)- Returns
GuardrailResult(passed=True)orGuardrailResult(passed=False, message="...")
Async Example
from connic import GuardrailResult
import httpx
import re
async def check(content: str, context: dict) -> GuardrailResult:
"""Check if the user has permission to use this agent.
Input guardrails run before middleware, so request data is read from
content. If you need middleware-enriched context, use an output guardrail
or enforce permissions in middleware instead.
"""
match = re.search(r'["\']user_id["\']\s*:\s*["\']([^"\']+)', content)
user_id = match.group(1) if match else None
if not user_id:
return GuardrailResult(passed=False, message="User not authenticated")
async with httpx.AsyncClient() as client:
resp = await client.get(f"https://api.example.com/users/{user_id}/permissions")
if resp.status_code != 200:
return GuardrailResult(passed=False, message="Could not verify permissions")
data = resp.json()
if not data.get("allowed"):
return GuardrailResult(passed=False, message="Insufficient permissions")
return GuardrailResult(passed=True)Logging from a custom guardrail
print, writes to sys.stderr, and stdlib logging calls from a custom guardrail are all captured per run and shown on the project Logs tab and in the run detail view, tagged with source guardrail.<name>. If the check function raises, the runtime catches the exception and auto-logs the traceback at error level under that same source. The exception is then handled as a guardrail violation according to the rule's mode: warn continues processing, while the default block mode stops processing and returns the configured rejection message.