Pydantic AI
Pydantic AI separates the model from the provider that carries it. That makes
the firewall a one-object change: build an OpenAIProvider pointed at the gateway,
hand it to the model, and every agent, tool, and output validator downstream is
firewalled without knowing it.
Install
pip install pydantic-ai
Configure
import os
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider
org = os.environ["VULNETIX_ORG_UUID"]
model = OpenAIChatModel(
"gpt-4o-mini",
provider=OpenAIProvider(
base_url=f"https://guardrails.vulnetix.com/openai/{org}/v1",
api_key=os.environ["VULNETIX_API_KEY"], # Vulnetix key, not an OpenAI key
),
)
agent = Agent(model, system_prompt="Be concise.")
result = agent.run_sync("Hello")
print(result.output)
Any OpenAI-compatible provider works through OpenAIProvider — change the slug in
the path to groq, mistral, deepseek, or openrouter, and pass that provider’s
own model string as the first argument. The gateway forwards it verbatim.
GroqProvider,
MistralProvider, and friends) when you want traffic firewalled: they carry their
own base URLs and their own credentials, and will talk to the provider directly.
OpenAIProvider with the gateway’s base_url is the one that goes through the
firewall.Streaming
Unchanged.
async with agent.run_stream("Hello") as result:
async for text in result.stream_text(delta=True):
print(text, end="", flush=True)
Handling a guardrail block
A request your policy refuses never reaches the provider. Pydantic AI wraps the
HTTP failure in pydantic_ai.exceptions.ModelHTTPError, which carries the status
code and the response body.
from pydantic_ai.exceptions import ModelHTTPError
try:
result = agent.run_sync(question)
except ModelHTTPError as e:
if e.status_code == 403:
error = (e.body or {}).get("error", {}) if isinstance(e.body, dict) else {}
code = error.get("code") # request_blocked, model_denied, ...
rule = error.get("blocked_by") # the rule that stopped it
raise RuntimeError(
f"AI firewall refused this agent run: {code} (rule: {rule})"
) from e
raise
code | Meaning |
|---|---|
request_blocked | A content guardrail matched. blocked_by names the rule; violations lists every rule that matched. |
provider_denied | Policy does not allow this provider. |
model_denied / model_not_allowed | Policy does not allow this model. |
provider_key_missing | No provider key in the vault for this provider. |
An agent run is a loop, and the block can land on any turn of it — including one
carrying a tool result the model asked for, not the prompt your user typed. Catch it
around run_sync / run, not around the first message.
pii_redact rule rewrites the
matched spans to the literal [REDACTED] and the run proceeds. With structured
outputs this is worth thinking about: the model is validating and populating your
Pydantic model from text it can no longer see in full, so a field that should have
been an email address may come back empty or invented. That is the guardrail working
as designed; the fix is in the policy, not the schema.Tool / function calling
@agent.tool functions pass through untouched — Pydantic AI drives the loop and the
gateway forwards it. Guardrails inspect each request the agent sends, and on the
turn after a tool runs, that request contains the tool’s return value. A tool that
reads a config file with credentials in it hands those credentials to the next
request, which the firewall can block or redact before the provider sees them.
Gotchas
api_keyis your Vulnetix key. A provider key here is the most common401.base_urlmust include the/v1suffix.- The string shorthand —
Agent("openai:gpt-4o-mini")— builds a default provider that talks straight to OpenAI. It cannot be firewalled. Construct the model and provider explicitly, as above, whenever you want the gateway in the path. ModelHTTPError.bodyis the parsed response body when the provider (here, the gateway) returned JSON, which it always does.