33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 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 99100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214

@dataclass(eq=False)classRunContextWrapper(Generic[TContext]):"""This wraps the context object that you passed to `Runner.run()`. It also contains    information about the usage of the agent run so far.    NOTE: Contexts are not passed to the LLM. They're a way to pass dependencies and data to code    you implement, like tool functions, callbacks, hooks, etc.    """context:TContext"""The context object (or None), passed by you to `Runner.run()`"""usage:Usage=field(default_factory=Usage)"""The usage of the agent run so far. For streamed responses, the usage will be stale until the    last chunk of the stream is processed.    """turn_input:list[TResponseInputItem]=field(default_factory=list)_approvals:dict[str,_ApprovalRecord]=field(default_factory=dict)tool_input:Any|None=None"""Structured input for the current agent tool run, when available."""@staticmethoddef_to_str_or_none(value:Any)->str|None:ifisinstance(value,str):returnvalueifvalueisnotNone:try:returnstr(value)exceptException:returnNonereturnNone@staticmethoddef_resolve_tool_name(approval_item:ToolApprovalItem)->str:raw=approval_item.raw_itemifapproval_item.tool_name:returnapproval_item.tool_namecandidate:Any|Noneifisinstance(raw,dict):candidate=raw.get("name")orraw.get("type")else:candidate=getattr(raw,"name",None)orgetattr(raw,"type",None)returnRunContextWrapper._to_str_or_none(candidate)or"unknown_tool"@staticmethoddef_resolve_call_id(approval_item:ToolApprovalItem)->str|None:raw=approval_item.raw_itemifisinstance(raw,dict):provider_data=raw.get("provider_data")if(isinstance(provider_data,dict)andprovider_data.get("type")=="mcp_approval_request"):candidate=provider_data.get("id")ifisinstance(candidate,str):returncandidatecandidate=raw.get("call_id")orraw.get("id")else:provider_data=getattr(raw,"provider_data",None)if(isinstance(provider_data,dict)andprovider_data.get("type")=="mcp_approval_request"):candidate=provider_data.get("id")ifisinstance(candidate,str):returncandidatecandidate=getattr(raw,"call_id",None)orgetattr(raw,"id",None)returnRunContextWrapper._to_str_or_none(candidate)def_get_or_create_approval_entry(self,tool_name:str)->_ApprovalRecord:approval_entry=self._approvals.get(tool_name)ifapproval_entryisNone:approval_entry=_ApprovalRecord()self._approvals[tool_name]=approval_entryreturnapproval_entrydefis_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.returnNonedef_apply_approval_decision(self,approval_item:ToolApprovalItem,*,always:bool,approve:bool)->None:"""Record an approval or rejection decision."""tool_name=self._resolve_tool_name(approval_item)call_id=self._resolve_call_id(approval_item)approval_entry=self._get_or_create_approval_entry(tool_name)ifalwaysorcall_idisNone:approval_entry.approved=approveapproval_entry.rejected=[]ifapproveelseTrueifnotapprove:approval_entry.approved=Falsereturnopposite=approval_entry.rejectedifapproveelseapproval_entry.approvedifisinstance(opposite,list)andcall_idinopposite:opposite.remove(call_id)target=approval_entry.approvedifapproveelseapproval_entry.rejectedifisinstance(target,list)andcall_idnotintarget:target.append(call_id)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,)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,)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)returnstatusdef_rebuild_approvals(self,approvals:dict[str,dict[str,Any]])->None:"""Restore approvals from serialized state."""self._approvals={}fortool_name,record_dictinapprovals.items():record=_ApprovalRecord()record.approved=record_dict.get("approved",[])record.rejected=record_dict.get("rejected",[])self._approvals[tool_name]=recorddef_fork_with_tool_input(self,tool_input:Any)->RunContextWrapper[TContext]:"""Create a child context that shares approvals and usage with tool input set."""fork=RunContextWrapper(context=self.context)fork.usage=self.usagefork._approvals=self._approvalsfork.turn_input=self.turn_inputfork.tool_input=tool_inputreturnforkdef_fork_without_tool_input(self)->RunContextWrapper[TContext]:"""Create a child context that shares approvals and usage without tool input."""fork=RunContextWrapper(context=self.context)fork.usage=self.usagefork._approvals=self._approvalsfork.turn_input=self.turn_inputreturnfork

217218219

@dataclass(eq=False)classAgentHookContext(RunContextWrapper[TContext]):"""Context passed to agent hooks (on_start, on_end)."""