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 99100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262

classMCPServer(abc.ABC):"""Base class for Model Context Protocol servers."""def__init__(self,use_structured_content:bool=False,require_approval:RequireApprovalSetting=None,failure_error_function:ToolErrorFunction|None|_UnsetType=_UNSET,tool_meta_resolver:MCPToolMetaResolver|None=None,):"""        Args:            use_structured_content: Whether to use `tool_result.structured_content` when calling an                MCP tool.Defaults to False for backwards compatibility - most MCP servers still                include the structured content in the `tool_result.content`, and using it by                default will cause duplicate content. You can set this to True if you know the                server will not duplicate the structured content in the `tool_result.content`.            require_approval: Approval policy for tools on this server. Accepts "always"/"never",                a dict of tool names to those values, a boolean, or an object with always/never                tool lists (mirroring TS requireApproval). Normalized into a needs_approval policy.            failure_error_function: Optional function used to convert MCP tool failures into                a model-visible error message. If explicitly set to None, tool errors will be                raised instead of converted. If left unset, the agent-level configuration (or                SDK default) will be used.            tool_meta_resolver: Optional callable that produces MCP request metadata (`_meta`) for                tool calls. It is invoked by the Agents SDK before calling `call_tool`.        """self.use_structured_content=use_structured_contentself._needs_approval_policy=self._normalize_needs_approval(require_approval=require_approval)self._failure_error_function=failure_error_functionself.tool_meta_resolver=tool_meta_resolver@abc.abstractmethodasyncdefconnect(self):"""Connect to the server. For example, this might mean spawning a subprocess or        opening a network connection. The server is expected to remain connected until        `cleanup()` is called.        """pass@property@abc.abstractmethoddefname(self)->str:"""A readable name for the server."""pass@abc.abstractmethodasyncdefcleanup(self):"""Cleanup the server. For example, this might mean closing a subprocess or        closing a network connection.        """pass@abc.abstractmethodasyncdeflist_tools(self,run_context:RunContextWrapper[Any]|None=None,agent:AgentBase|None=None,)->list[MCPTool]:"""List the tools available on the server."""pass@abc.abstractmethodasyncdefcall_tool(self,tool_name:str,arguments:dict[str,Any]|None,meta:dict[str,Any]|None=None,)->CallToolResult:"""Invoke a tool on the server."""pass@propertydefcached_tools(self)->list[MCPTool]|None:"""Return the most recently fetched tools list, if available.        Implementations may return `None` when tools have not been fetched yet or caching is        disabled.        """returnNone@abc.abstractmethodasyncdeflist_prompts(self,)->ListPromptsResult:"""List the prompts available on the server."""pass@abc.abstractmethodasyncdefget_prompt(self,name:str,arguments:dict[str,Any]|None=None)->GetPromptResult:"""Get a specific prompt from the server."""pass@staticmethoddef_normalize_needs_approval(*,require_approval:RequireApprovalSetting,)->(bool|dict[str,bool]|Callable[[RunContextWrapper[Any],AgentBase,MCPTool],MaybeAwaitable[bool]]):"""Normalize approval inputs to booleans or a name->bool map."""ifrequire_approvalisNone:returnFalsedef_to_bool(value:str)->bool:returnvalue=="always"def_is_tool_list_schema(value:object)->bool:ifnotisinstance(value,dict):returnFalseforkeyin("always","never"):ifkeynotinvalue:continueentry=value.get(key)ifisinstance(entry,dict)and"tool_names"inentry:returnTruereturnFalseifisinstance(require_approval,dict)and_is_tool_list_schema(require_approval):always_entry:RequireApprovalToolList|Any=require_approval.get("always",{})never_entry:RequireApprovalToolList|Any=require_approval.get("never",{})always_names=(always_entry.get("tool_names",[])ifisinstance(always_entry,dict)else[])never_names=never_entry.get("tool_names",[])ifisinstance(never_entry,dict)else[]tool_list_mapping:dict[str,bool]={}fornameinalways_names:tool_list_mapping[str(name)]=Truefornameinnever_names:tool_list_mapping[str(name)]=Falsereturntool_list_mappingifisinstance(require_approval,dict):tool_mapping:dict[str,bool]={}forname,valueinrequire_approval.items():ifisinstance(value,bool):tool_mapping[str(name)]=valueelifisinstance(value,str)andvaluein("always","never"):tool_mapping[str(name)]=_to_bool(value)returntool_mappingifisinstance(require_approval,bool):returnrequire_approvalreturn_to_bool(require_approval)def_get_needs_approval_for_tool(self,tool:MCPTool,agent:AgentBase|None,)->bool|Callable[[RunContextWrapper[Any],dict[str,Any],str],Awaitable[bool]]:"""Return a FunctionTool.needs_approval value for a given MCP tool."""policy=self._needs_approval_policyifcallable(policy):ifagentisNone:returnFalseasyncdef_needs_approval(run_context:RunContextWrapper[Any],_args:dict[str,Any],_call_id:str)->bool:result=policy(run_context,agent,tool)ifinspect.isawaitable(result):result=awaitresultreturnbool(result)return_needs_approvalifisinstance(policy,dict):returnbool(policy.get(tool.name,False))returnbool(policy)def_get_failure_error_function(self,agent_failure_error_function:ToolErrorFunction|None)->ToolErrorFunction|None:"""Return the effective error handler for MCP tool failures."""ifself._failure_error_functionis_UNSET:returnagent_failure_error_functionreturncast(ToolErrorFunction|None,self._failure_error_function)

715716717718719720721722723724725726727728729730731732733734735736737738739740741

classMCPServerStdioParams(TypedDict):"""Mirrors `mcp.client.stdio.StdioServerParameters`, but lets you pass params without another    import.    """command:str"""The executable to run to start the server. For example, `python` or `node`."""args:NotRequired[list[str]]"""Command line args to pass to the `command` executable. For example, `['foo.py']` or    `['server.js', '--port', '8080']`."""env:NotRequired[dict[str,str]]"""The environment variables to set for the server. ."""cwd:NotRequired[str|Path]"""The working directory to use when spawning the process."""encoding:NotRequired[str]"""The text encoding used when sending/receiving messages to the server. Defaults to `utf-8`."""encoding_error_handler:NotRequired[Literal["strict","ignore","replace"]]"""The text encoding error handler. Defaults to `strict`.    See https://docs.python.org/3/library/codecs.html#codec-base-classes for    explanations of possible values.    """

744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841

classMCPServerStdio(_MCPServerWithClientSession):"""MCP server implementation that uses the stdio transport. See the [spec]    (https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#stdio) for    details.    """def__init__(self,params:MCPServerStdioParams,cache_tools_list:bool=False,name:str|None=None,client_session_timeout_seconds:float|None=5,tool_filter:ToolFilter=None,use_structured_content:bool=False,max_retry_attempts:int=0,retry_backoff_seconds_base:float=1.0,message_handler:MessageHandlerFnT|None=None,require_approval:RequireApprovalSetting=None,failure_error_function:ToolErrorFunction|None|_UnsetType=_UNSET,tool_meta_resolver:MCPToolMetaResolver|None=None,):"""Create a new MCP server based on the stdio transport.        Args:            params: The params that configure the server. This includes the command to run to                start the server, the args to pass to the command, the environment variables to                set for the server, the working directory to use when spawning the process, and                the text encoding used when sending/receiving messages to the server.            cache_tools_list: Whether to cache the tools list. If `True`, the tools list will be                cached and only fetched from the server once. If `False`, the tools list will be                fetched from the server on each call to `list_tools()`. The cache can be                invalidated by calling `invalidate_tools_cache()`. You should set this to `True`                if you know the server will not change its tools list, because it can drastically                improve latency (by avoiding a round-trip to the server every time).            name: A readable name for the server. If not provided, we'll create one from the                command.            client_session_timeout_seconds: the read timeout passed to the MCP ClientSession.            tool_filter: The tool filter to use for filtering tools.            use_structured_content: Whether to use `tool_result.structured_content` when calling an                MCP tool. Defaults to False for backwards compatibility - most MCP servers still                include the structured content in the `tool_result.content`, and using it by                default will cause duplicate content. You can set this to True if you know the                server will not duplicate the structured content in the `tool_result.content`.            max_retry_attempts: Number of times to retry failed list_tools/call_tool calls.                Defaults to no retries.            retry_backoff_seconds_base: The base delay, in seconds, for exponential                backoff between retries.            message_handler: Optional handler invoked for session messages as delivered by the                ClientSession.            require_approval: Approval policy for tools on this server. Accepts "always"/"never",                a dict of tool names to those values, or an object with always/never tool lists.            failure_error_function: Optional function used to convert MCP tool failures into                a model-visible error message. If explicitly set to None, tool errors will be                raised instead of converted. If left unset, the agent-level configuration (or                SDK default) will be used.            tool_meta_resolver: Optional callable that produces MCP request metadata (`_meta`) for                tool calls. It is invoked by the Agents SDK before calling `call_tool`.        """super().__init__(cache_tools_list=cache_tools_list,client_session_timeout_seconds=client_session_timeout_seconds,tool_filter=tool_filter,use_structured_content=use_structured_content,max_retry_attempts=max_retry_attempts,retry_backoff_seconds_base=retry_backoff_seconds_base,message_handler=message_handler,require_approval=require_approval,failure_error_function=failure_error_function,tool_meta_resolver=tool_meta_resolver,)self.params=StdioServerParameters(command=params["command"],args=params.get("args",[]),env=params.get("env"),cwd=params.get("cwd"),encoding=params.get("encoding","utf-8"),encoding_error_handler=params.get("encoding_error_handler","strict"),)self._name=nameorf"stdio:{self.params.command}"defcreate_streams(self,)->AbstractAsyncContextManager[tuple[MemoryObjectReceiveStream[SessionMessage|Exception],MemoryObjectSendStream[SessionMessage],GetSessionIdCallback|None,]]:"""Create the streams for the server."""returnstdio_client(self.params)@propertydefname(self)->str:"""A readable name for the server."""returnself._name

844845846847848849850851852853854855856857

classMCPServerSseParams(TypedDict):"""Mirrors the params in`mcp.client.sse.sse_client`."""url:str"""The URL of the server."""headers:NotRequired[dict[str,str]]"""The headers to send to the server."""timeout:NotRequired[float]"""The timeout for the HTTP request. Defaults to 5 seconds."""sse_read_timeout:NotRequired[float]"""The timeout for the SSE connection, in seconds. Defaults to 5 minutes."""

860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956

classMCPServerSse(_MCPServerWithClientSession):"""MCP server implementation that uses the HTTP with SSE transport. See the [spec]    (https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/transports/#http-with-sse)    for details.    """def__init__(self,params:MCPServerSseParams,cache_tools_list:bool=False,name:str|None=None,client_session_timeout_seconds:float|None=5,tool_filter:ToolFilter=None,use_structured_content:bool=False,max_retry_attempts:int=0,retry_backoff_seconds_base:float=1.0,message_handler:MessageHandlerFnT|None=None,require_approval:RequireApprovalSetting=None,failure_error_function:ToolErrorFunction|None|_UnsetType=_UNSET,tool_meta_resolver:MCPToolMetaResolver|None=None,):"""Create a new MCP server based on the HTTP with SSE transport.        Args:            params: The params that configure the server. This includes the URL of the server,                the headers to send to the server, the timeout for the HTTP request, and the                timeout for the SSE connection.            cache_tools_list: Whether to cache the tools list. If `True`, the tools list will be                cached and only fetched from the server once. If `False`, the tools list will be                fetched from the server on each call to `list_tools()`. The cache can be                invalidated by calling `invalidate_tools_cache()`. You should set this to `True`                if you know the server will not change its tools list, because it can drastically                improve latency (by avoiding a round-trip to the server every time).            name: A readable name for the server. If not provided, we'll create one from the                URL.            client_session_timeout_seconds: the read timeout passed to the MCP ClientSession.            tool_filter: The tool filter to use for filtering tools.            use_structured_content: Whether to use `tool_result.structured_content` when calling an                MCP tool. Defaults to False for backwards compatibility - most MCP servers still                include the structured content in the `tool_result.content`, and using it by                default will cause duplicate content. You can set this to True if you know the                server will not duplicate the structured content in the `tool_result.content`.            max_retry_attempts: Number of times to retry failed list_tools/call_tool calls.                Defaults to no retries.            retry_backoff_seconds_base: The base delay, in seconds, for exponential                backoff between retries.            message_handler: Optional handler invoked for session messages as delivered by the                ClientSession.            require_approval: Approval policy for tools on this server. Accepts "always"/"never",                a dict of tool names to those values, or an object with always/never tool lists.            failure_error_function: Optional function used to convert MCP tool failures into                a model-visible error message. If explicitly set to None, tool errors will be                raised instead of converted. If left unset, the agent-level configuration (or                SDK default) will be used.            tool_meta_resolver: Optional callable that produces MCP request metadata (`_meta`) for                tool calls. It is invoked by the Agents SDK before calling `call_tool`.        """super().__init__(cache_tools_list=cache_tools_list,client_session_timeout_seconds=client_session_timeout_seconds,tool_filter=tool_filter,use_structured_content=use_structured_content,max_retry_attempts=max_retry_attempts,retry_backoff_seconds_base=retry_backoff_seconds_base,message_handler=message_handler,require_approval=require_approval,failure_error_function=failure_error_function,tool_meta_resolver=tool_meta_resolver,)self.params=paramsself._name=nameorf"sse:{self.params['url']}"defcreate_streams(self,)->AbstractAsyncContextManager[tuple[MemoryObjectReceiveStream[SessionMessage|Exception],MemoryObjectSendStream[SessionMessage],GetSessionIdCallback|None,]]:"""Create the streams for the server."""returnsse_client(url=self.params["url"],headers=self.params.get("headers",None),timeout=self.params.get("timeout",5),sse_read_timeout=self.params.get("sse_read_timeout",60*5),)@propertydefname(self)->str:"""A readable name for the server."""returnself._name