Two calls around your model
Screen the input, call your model, screen the response. The 4-tuple (tenant, app, agent, env) resolves which policy applies, so no policy id travels with the call.
scope = dict(
tenant_id=ORG,
app_id=PROJECT,
agent_id="triage",
env="production",
)
inp = guardrails.evaluate_input(
prompt, **scope
)
if inp.is_blocked:
return inp.user_message
prompt = inp.sanitized_text or prompt
answer = llm.complete(prompt)
out = guardrails.evaluate_output(
answer, **scope
)
return out.sanitized_text or answerStreaming
Post response chunks and the engine evaluates a rolling window (200 chars, 40 overlap by default), streaming verdicts back over SSE. Treat a block event as terminal. Streaming is on the async client, and the stream resolves its policy from the bundle rather than the call.
stream = guardrails.evaluate_stream(
chunks=chunks
)
async for ev in stream:
if ev.event == "block":
break
if ev.event == "chunk":
yield ev.data["text"]Tool-call gating
Gate a tool call before it executes, from whatever hook your framework already gives you. Plan, step, retrieval, and memory writes follow the same pattern against their own endpoints.
res = guardrails.evaluate_tool(
tool_name, tool_args, **scope
)
if res.is_blocked:
raise ToolBlocked(
res.user_message
)
tool_args = (
res.sanitized_tool_args
or tool_args
)