474849505152535455565758596061626364656667686970717273747576777879808182

def__init__(self,context:TContext,usage:Usage|object=_MISSING,tool_name:str|object=_MISSING,tool_call_id:str|object=_MISSING,tool_arguments:str|object=_MISSING,tool_call:ResponseFunctionToolCall|None=None,*,turn_input:list[TResponseInputItem]|None=None,_approvals:dict[str,_ApprovalRecord]|None=None,tool_input:Any|None=None,)->None:"""Preserve the v0.7 positional constructor while accepting new context fields."""resolved_usage=Usage()ifusageis_MISSINGelsecast(Usage,usage)super().__init__(context=context,usage=resolved_usage,turn_input=list(turn_inputor[]),_approvals={}if_approvalsisNoneelse_approvals,tool_input=tool_input,)self.tool_name=(_assert_must_pass_tool_name()iftool_nameis_MISSINGelsecast(str,tool_name))self.tool_arguments=(_assert_must_pass_tool_arguments()iftool_argumentsis_MISSINGelsecast(str,tool_arguments))self.tool_call_id=(_assert_must_pass_tool_call_id()iftool_call_idis_MISSINGelsecast(str,tool_call_id))self.tool_call=tool_call

 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99100101102103104105106107108109110

@classmethoddeffrom_agent_context(cls,context:RunContextWrapper[TContext],tool_call_id:str,tool_call:ResponseFunctionToolCall|None=None,)->ToolContext:"""    Create a ToolContext from a RunContextWrapper.    """# Grab the names of the RunContextWrapper's init=True fieldsbase_values:dict[str,Any]={f.name:getattr(context,f.name)forfinfields(RunContextWrapper)iff.init}tool_name=tool_call.nameiftool_callisnotNoneelse_assert_must_pass_tool_name()tool_args=(tool_call.argumentsiftool_callisnotNoneelse_assert_must_pass_tool_arguments())tool_context=cls(tool_name=tool_name,tool_call_id=tool_call_id,tool_arguments=tool_args,tool_call=tool_call,**base_values,)returntool_context

110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139

defis_tool_approved(self,tool_name:str,call_id:str)->bool|None:"""Return True/False/None for the given tool call."""approval_entry=self._approvals.get(tool_name)ifnotapproval_entry:returnNone# Check for permanent approval/rejectionifapproval_entry.approvedisTrueandapproval_entry.rejectedisTrue:# Approval takes precedencereturnTrueifapproval_entry.approvedisTrue:returnTrueifapproval_entry.rejectedisTrue:returnFalseapproved_ids=(set(approval_entry.approved)ifisinstance(approval_entry.approved,list)elseset())rejected_ids=(set(approval_entry.rejected)ifisinstance(approval_entry.rejected,list)elseset())ifcall_idinapproved_ids:returnTrueifcall_idinrejected_ids:returnFalse# Per-call approvals are scoped to the exact call ID, so other calls require a new decision.returnNone

164165166167168169170

defapprove_tool(self,approval_item:ToolApprovalItem,always_approve:bool=False)->None:"""Approve a tool call, optionally for all future calls."""self._apply_approval_decision(approval_item,always=always_approve,approve=True,)

172173174175176177178

defreject_tool(self,approval_item:ToolApprovalItem,always_reject:bool=False)->None:"""Reject a tool call, optionally for all future calls."""self._apply_approval_decision(approval_item,always=always_reject,approve=False,)

180181182183184185186187188

defget_approval_status(self,tool_name:str,call_id:str,*,existing_pending:ToolApprovalItem|None=None)->bool|None:"""Return approval status, retrying with pending item's tool name if necessary."""status=self.is_tool_approved(tool_name,call_id)ifstatusisNoneandexisting_pending:fallback_tool_name=self._resolve_tool_name(existing_pending)status=self.is_tool_approved(fallback_tool_name,call_id)returnstatus