1920212223242526272829303132

@dataclassclassGuardrailFunctionOutput:"""The output of a guardrail function."""output_info:Any"""    Optional information about the guardrail's output. For example, the guardrail could include    information about the checks it performed and granular results.    """tripwire_triggered:bool"""    Whether the tripwire was triggered. If triggered, the agent's execution will be halted.    """

3536373839404142434445

@dataclassclassInputGuardrailResult:"""The result of a guardrail run."""guardrail:InputGuardrail[Any]"""    The guardrail that was run.    """output:GuardrailFunctionOutput"""The output of the guardrail function."""

484950515253545556575859606162636465666768

@dataclassclassOutputGuardrailResult:"""The result of a guardrail run."""guardrail:OutputGuardrail[Any]"""    The guardrail that was run.    """agent_output:Any"""    The output of the agent that was checked by the guardrail.    """agent:Agent[Any]"""    The agent that was checked by the guardrail.    """output:GuardrailFunctionOutput"""The output of the guardrail function."""

 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130

@dataclassclassInputGuardrail(Generic[TContext]):"""Input guardrails are checks that run either in parallel with the agent or before it starts.    They can be used to do things like:    - Check if input messages are off-topic    - Take over control of the agent's execution if an unexpected input is detected    You can use the `@input_guardrail()` decorator to turn a function into an `InputGuardrail`, or    create an `InputGuardrail` manually.    Guardrails return a `GuardrailResult`. If `result.tripwire_triggered` is `True`,    the agent's execution will immediately stop, and    an `InputGuardrailTripwireTriggered` exception will be raised    """guardrail_function:Callable[[RunContextWrapper[TContext],Agent[Any],str|list[TResponseInputItem]],MaybeAwaitable[GuardrailFunctionOutput],]"""A function that receives the agent input and the context, and returns a     `GuardrailResult`. The result marks whether the tripwire was triggered, and can optionally     include information about the guardrail's output.    """name:str|None=None"""The name of the guardrail, used for tracing. If not provided, we'll use the guardrail    function's name.    """run_in_parallel:bool=True"""Whether the guardrail runs concurrently with the agent (True, default) or before    the agent starts (False).    """defget_name(self)->str:ifself.name:returnself.namereturnself.guardrail_function.__name__asyncdefrun(self,agent:Agent[Any],input:str|list[TResponseInputItem],context:RunContextWrapper[TContext],)->InputGuardrailResult:ifnotcallable(self.guardrail_function):raiseUserError(f"Guardrail function must be callable, got{self.guardrail_function}")output=self.guardrail_function(context,agent,input)ifinspect.isawaitable(output):returnInputGuardrailResult(guardrail=self,output=awaitoutput,)returnInputGuardrailResult(guardrail=self,output=output,)

133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185

@dataclassclassOutputGuardrail(Generic[TContext]):"""Output guardrails are checks that run on the final output of an agent.    They can be used to do check if the output passes certain validation criteria    You can use the `@output_guardrail()` decorator to turn a function into an `OutputGuardrail`,    or create an `OutputGuardrail` manually.    Guardrails return a `GuardrailResult`. If `result.tripwire_triggered` is `True`, an    `OutputGuardrailTripwireTriggered` exception will be raised.    """guardrail_function:Callable[[RunContextWrapper[TContext],Agent[Any],Any],MaybeAwaitable[GuardrailFunctionOutput],]"""A function that receives the final agent, its output, and the context, and returns a     `GuardrailResult`. The result marks whether the tripwire was triggered, and can optionally     include information about the guardrail's output.    """name:str|None=None"""The name of the guardrail, used for tracing. If not provided, we'll use the guardrail    function's name.    """defget_name(self)->str:ifself.name:returnself.namereturnself.guardrail_function.__name__asyncdefrun(self,context:RunContextWrapper[TContext],agent:Agent[Any],agent_output:Any)->OutputGuardrailResult:ifnotcallable(self.guardrail_function):raiseUserError(f"Guardrail function must be callable, got{self.guardrail_function}")output=self.guardrail_function(context,agent,agent_output)ifinspect.isawaitable(output):returnOutputGuardrailResult(guardrail=self,agent=agent,agent_output=agent_output,output=awaitoutput,)returnOutputGuardrailResult(guardrail=self,agent=agent,agent_output=agent_output,output=output,)

224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270

definput_guardrail(func:_InputGuardrailFuncSync[TContext_co]|_InputGuardrailFuncAsync[TContext_co]|None=None,*,name:str|None=None,run_in_parallel:bool=True,)->(InputGuardrail[TContext_co]|Callable[[_InputGuardrailFuncSync[TContext_co]|_InputGuardrailFuncAsync[TContext_co]],InputGuardrail[TContext_co],]):"""    Decorator that transforms a sync or async function into an `InputGuardrail`.    It can be used directly (no parentheses) or with keyword args, e.g.:        @input_guardrail        def my_sync_guardrail(...): ...        @input_guardrail(name="guardrail_name", run_in_parallel=False)        async def my_async_guardrail(...): ...    Args:        func: The guardrail function to wrap.        name: Optional name for the guardrail. If not provided, uses the function's name.        run_in_parallel: Whether to run the guardrail concurrently with the agent (True, default)            or before the agent starts (False).    """defdecorator(f:_InputGuardrailFuncSync[TContext_co]|_InputGuardrailFuncAsync[TContext_co],)->InputGuardrail[TContext_co]:returnInputGuardrail(guardrail_function=f,# If not set, guardrail name uses the function’s name by default.name=nameifnameelsef.__name__,run_in_parallel=run_in_parallel,)iffuncisnotNone:# Decorator was used without parenthesesreturndecorator(func)# Decorator used with keyword argumentsreturndecorator

305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343

defoutput_guardrail(func:_OutputGuardrailFuncSync[TContext_co]|_OutputGuardrailFuncAsync[TContext_co]|None=None,*,name:str|None=None,)->(OutputGuardrail[TContext_co]|Callable[[_OutputGuardrailFuncSync[TContext_co]|_OutputGuardrailFuncAsync[TContext_co]],OutputGuardrail[TContext_co],]):"""    Decorator that transforms a sync or async function into an `OutputGuardrail`.    It can be used directly (no parentheses) or with keyword args, e.g.:        @output_guardrail        def my_sync_guardrail(...): ...        @output_guardrail(name="guardrail_name")        async def my_async_guardrail(...): ...    """defdecorator(f:_OutputGuardrailFuncSync[TContext_co]|_OutputGuardrailFuncAsync[TContext_co],)->OutputGuardrail[TContext_co]:returnOutputGuardrail(guardrail_function=f,# Guardrail name defaults to function's name when not specified (None).name=nameifnameelsef.__name__,)iffuncisnotNone:# Decorator was used without parenthesesreturndecorator(func)# Decorator used with keyword argumentsreturndecorator