3233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293

classMCPServer(abc.ABC):"""Base class for Model Context Protocol servers."""def__init__(self,use_structured_content:bool=False):"""        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`.        """self.use_structured_content=use_structured_content@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)->CallToolResult:"""Invoke a tool on the server."""pass@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

358359360361362363364365366367368369370371372373374375376377378379380381382383384

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.    """

387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470

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,):"""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.        """super().__init__(cache_tools_list,client_session_timeout_seconds,tool_filter,use_structured_content,max_retry_attempts,retry_backoff_seconds_base,message_handler=message_handler,)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

473474475476477478479480481482483484485486

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."""

489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571

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,):"""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.        """super().__init__(cache_tools_list,client_session_timeout_seconds,tool_filter,use_structured_content,max_retry_attempts,retry_backoff_seconds_base,message_handler=message_handler,)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

574575576577578579580581582583584585586587588589590591592593

classMCPServerStreamableHttpParams(TypedDict):"""Mirrors the params in`mcp.client.streamable_http.streamablehttp_client`."""url:str"""The URL of the server."""headers:NotRequired[dict[str,str]]"""The headers to send to the server."""timeout:NotRequired[timedelta|float]"""The timeout for the HTTP request. Defaults to 5 seconds."""sse_read_timeout:NotRequired[timedelta|float]"""The timeout for the SSE connection, in seconds. Defaults to 5 minutes."""terminate_on_close:NotRequired[bool]"""Terminate on close"""httpx_client_factory:NotRequired[HttpClientFactory]"""Custom HTTP client factory for configuring httpx.AsyncClient behavior."""

596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691

classMCPServerStreamableHttp(_MCPServerWithClientSession):"""MCP server implementation that uses the Streamable HTTP transport. See the [spec]    (https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http)    for details.    """def__init__(self,params:MCPServerStreamableHttpParams,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,):"""Create a new MCP server based on the Streamable HTTP 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, the                timeout for the Streamable HTTP connection, whether we need to                terminate on close, and an optional custom HTTP client factory.            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.        """super().__init__(cache_tools_list,client_session_timeout_seconds,tool_filter,use_structured_content,max_retry_attempts,retry_backoff_seconds_base,message_handler=message_handler,)self.params=paramsself._name=nameorf"streamable_http:{self.params['url']}"defcreate_streams(self,)->AbstractAsyncContextManager[tuple[MemoryObjectReceiveStream[SessionMessage|Exception],MemoryObjectSendStream[SessionMessage],GetSessionIdCallback|None,]]:"""Create the streams for the server."""# Only pass httpx_client_factory if it's providedif"httpx_client_factory"inself.params:returnstreamablehttp_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),terminate_on_close=self.params.get("terminate_on_close",True),httpx_client_factory=self.params["httpx_client_factory"],)else:returnstreamablehttp_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),terminate_on_close=self.params.get("terminate_on_close",True),)@propertydefname(self)->str:"""A readable name for the server."""returnself._name