@input_guardrail
async def rate_info_guardrail(
ctx: RunContextWrapper[OrchestratorContext], # your context type
agent: Agent,
input: str | list[TResponseInputItem], # same sig the SDK expects
) -> GuardrailFunctionOutput:
"""
Abort the run if any critical rate-info fields are missing, or if
is_bid_request_sent is still False.
"""
_REQUIRED_RATE_FIELDS = [
"maximum_rate",
"minimum_rate",
"rate_usd",
"is_bid_request_sent",
]
rate = ctx.context.load_context.rate_info # <-- your own object
# 2️⃣ Collect the names of missing / invalid fields.
missing: list[str] = []
for name in _REQUIRED_RATE_FIELDS:
value = getattr(rate, name, None)
if value is None: # not set
missing.append(name)
elif name == "is_bid_request_sent" and value is False:
missing.append(name)
# 3️⃣ Decide whether to trip the wire.
trip = bool(missing)
print("Rate info guardrail triggered...")
# 4️⃣ Return the standard GuardrailFunctionOutput.
return GuardrailFunctionOutput(
output_info={"missing_fields": missing}, # great for debugging/logging
tripwire_triggered=trip,
)
negotiation_agent = Agent[OrchestratorContext](
name="Negotiation Agent",
instructions=dynamic_negotiation_agent_instructions,
model=OpenAIChatCompletionsModel(
model=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"), openai_client=azure_client
),
tools=[
FunctionTool(
name="send_reply",
on_invoke_tool=send_reply,
description="This tool will send the HTML email body to the broker",
params_json_schema=AgentsReq.model_json_schema(),
),
],
input_guardrails=[rate_info_guardrail],
)
In my logs I see that my negotiation_agent is running, but I don't see the
print("Rate info guardrail triggered...")