API Reference

class ClientSession( BaseSession[ types.ClientRequest, types.ClientNotification, types.ClientResult, types.ServerRequest, types.ServerNotification, ] ): def __init__( self, read_stream: MemoryObjectReceiveStream[SessionMessage | Exception], write_stream: MemoryObjectSendStream[SessionMessage], read_timeout_seconds: timedelta | None = None, sampling_callback: SamplingFnT | None = None, elicitation_callback: ElicitationFnT | None = None, list_roots_callback: ListRootsFnT | None = None, logging_callback: LoggingFnT | None = None, message_handler: MessageHandlerFnT | None = None, client_info: types.Implementation | None = None, *, sampling_capabilities: types.SamplingCapability | None = None, experimental_task_handlers: ExperimentalTaskHandlers | None = None, ) -> None: super().__init__( read_stream, write_stream, types.ServerRequest, types.ServerNotification, read_timeout_seconds=read_timeout_seconds, ) self._client_info = client_info or DEFAULT_CLIENT_INFO self._sampling_callback = sampling_callback or _default_sampling_callback self._sampling_capabilities = sampling_capabilities self._elicitation_callback = elicitation_callback or _default_elicitation_callback self._list_roots_callback = list_roots_callback or _default_list_roots_callback self._logging_callback = logging_callback or _default_logging_callback self._message_handler = message_handler or _default_message_handler self._tool_output_schemas: dict[str, dict[str, Any] | None] = {} self._server_capabilities: types.ServerCapabilities | None = None self._experimental_features: ExperimentalClientFeatures | None = None # Experimental: Task handlers (use defaults if not provided) self._task_handlers = experimental_task_handlers or ExperimentalTaskHandlers() async def initialize(self) -> types.InitializeResult: sampling = ( (self._sampling_capabilities or types.SamplingCapability()) if self._sampling_callback is not _default_sampling_callback else None ) elicitation = ( types.ElicitationCapability( form=types.FormElicitationCapability(), url=types.UrlElicitationCapability(), ) if self._elicitation_callback is not _default_elicitation_callback else None ) roots = ( # TODO: Should this be based on whether we # _will_ send notifications, or only whether # they're supported? types.RootsCapability(listChanged=True) if self._list_roots_callback is not _default_list_roots_callback else None ) result = await self.send_request( types.ClientRequest( types.InitializeRequest( params=types.InitializeRequestParams( protocolVersion=types.LATEST_PROTOCOL_VERSION, capabilities=types.ClientCapabilities( sampling=sampling, elicitation=elicitation, experimental=None, roots=roots, tasks=self._task_handlers.build_capability(), ), clientInfo=self._client_info, ), ) ), types.InitializeResult, ) if result.protocolVersion not in SUPPORTED_PROTOCOL_VERSIONS: raise RuntimeError(f"Unsupported protocol version from the server: {result.protocolVersion}") self._server_capabilities = result.capabilities await self.send_notification(types.ClientNotification(types.InitializedNotification())) return result def get_server_capabilities(self) -> types.ServerCapabilities | None: """Return the server capabilities received during initialization. Returns None if the session has not been initialized yet. """ return self._server_capabilities @property def experimental(self) -> ExperimentalClientFeatures: """Experimental APIs for tasks and other features. WARNING: These APIs are experimental and may change without notice. Example: status = await session.experimental.get_task(task_id) result = await session.experimental.get_task_result(task_id, CallToolResult) """ if self._experimental_features is None: self._experimental_features = ExperimentalClientFeatures(self) return self._experimental_features async def send_ping(self) -> types.EmptyResult: """Send a ping request.""" return await self.send_request( types.ClientRequest(types.PingRequest()), types.EmptyResult, ) async def send_progress_notification( self, progress_token: str | int, progress: float, total: float | None = None, message: str | None = None, ) -> None: """Send a progress notification.""" await self.send_notification( types.ClientNotification( types.ProgressNotification( params=types.ProgressNotificationParams( progressToken=progress_token, progress=progress, total=total, message=message, ), ), ) ) async def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResult: """Send a logging/setLevel request.""" return await self.send_request( # pragma: no cover types.ClientRequest( types.SetLevelRequest( params=types.SetLevelRequestParams(level=level), ) ), types.EmptyResult, ) @overload @deprecated("Use list_resources(params=PaginatedRequestParams(...)) instead") async def list_resources(self, cursor: str | None) -> types.ListResourcesResult: ... @overload async def list_resources(self, *, params: types.PaginatedRequestParams | None) -> types.ListResourcesResult: ... @overload async def list_resources(self) -> types.ListResourcesResult: ... async def list_resources( self, cursor: str | None = None, *, params: types.PaginatedRequestParams | None = None, ) -> types.ListResourcesResult: """Send a resources/list request. Args: cursor: Simple cursor string for pagination (deprecated, use params instead) params: Full pagination parameters including cursor and any future fields """ if params is not None and cursor is not None: raise ValueError("Cannot specify both cursor and params") if params is not None: request_params = params elif cursor is not None: request_params = types.PaginatedRequestParams(cursor=cursor) else: request_params = None return await self.send_request( types.ClientRequest(types.ListResourcesRequest(params=request_params)), types.ListResourcesResult, ) @overload @deprecated("Use list_resource_templates(params=PaginatedRequestParams(...)) instead") async def list_resource_templates(self, cursor: str | None) -> types.ListResourceTemplatesResult: ... @overload async def list_resource_templates( self, *, params: types.PaginatedRequestParams | None ) -> types.ListResourceTemplatesResult: ... @overload async def list_resource_templates(self) -> types.ListResourceTemplatesResult: ... async def list_resource_templates( self, cursor: str | None = None, *, params: types.PaginatedRequestParams | None = None, ) -> types.ListResourceTemplatesResult: """Send a resources/templates/list request. Args: cursor: Simple cursor string for pagination (deprecated, use params instead) params: Full pagination parameters including cursor and any future fields """ if params is not None and cursor is not None: raise ValueError("Cannot specify both cursor and params") if params is not None: request_params = params elif cursor is not None: request_params = types.PaginatedRequestParams(cursor=cursor) else: request_params = None return await self.send_request( types.ClientRequest(types.ListResourceTemplatesRequest(params=request_params)), types.ListResourceTemplatesResult, ) async def read_resource(self, uri: AnyUrl) -> types.ReadResourceResult: """Send a resources/read request.""" return await self.send_request( types.ClientRequest( types.ReadResourceRequest( params=types.ReadResourceRequestParams(uri=uri), ) ), types.ReadResourceResult, ) async def subscribe_resource(self, uri: AnyUrl) -> types.EmptyResult: """Send a resources/subscribe request.""" return await self.send_request( # pragma: no cover types.ClientRequest( types.SubscribeRequest( params=types.SubscribeRequestParams(uri=uri), ) ), types.EmptyResult, ) async def unsubscribe_resource(self, uri: AnyUrl) -> types.EmptyResult: """Send a resources/unsubscribe request.""" return await self.send_request( # pragma: no cover types.ClientRequest( types.UnsubscribeRequest( params=types.UnsubscribeRequestParams(uri=uri), ) ), types.EmptyResult, ) async def call_tool( self, name: str, arguments: dict[str, Any] | None = None, read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, *, meta: dict[str, Any] | None = None, ) -> types.CallToolResult: """Send a tools/call request with optional progress callback support.""" _meta: types.RequestParams.Meta | None = None if meta is not None: _meta = types.RequestParams.Meta(**meta) result = await self.send_request( types.ClientRequest( types.CallToolRequest( params=types.CallToolRequestParams(name=name, arguments=arguments, _meta=_meta), ) ), types.CallToolResult, request_read_timeout_seconds=read_timeout_seconds, progress_callback=progress_callback, ) if not result.isError: await self._validate_tool_result(name, result) return result async def _validate_tool_result(self, name: str, result: types.CallToolResult) -> None: """Validate the structured content of a tool result against its output schema.""" if name not in self._tool_output_schemas: # refresh output schema cache await self.list_tools() output_schema = None if name in self._tool_output_schemas: output_schema = self._tool_output_schemas.get(name) else: logger.warning(f"Tool {name} not listed by server, cannot validate any structured content") if output_schema is not None: from jsonschema import SchemaError, ValidationError, validate if result.structuredContent is None: raise RuntimeError( f"Tool {name} has an output schema but did not return structured content" ) # pragma: no cover try: validate(result.structuredContent, output_schema) except ValidationError as e: raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}") # pragma: no cover except SchemaError as e: # pragma: no cover raise RuntimeError(f"Invalid schema for tool {name}: {e}") # pragma: no cover @overload @deprecated("Use list_prompts(params=PaginatedRequestParams(...)) instead") async def list_prompts(self, cursor: str | None) -> types.ListPromptsResult: ... @overload async def list_prompts(self, *, params: types.PaginatedRequestParams | None) -> types.ListPromptsResult: ... @overload async def list_prompts(self) -> types.ListPromptsResult: ... async def list_prompts( self, cursor: str | None = None, *, params: types.PaginatedRequestParams | None = None, ) -> types.ListPromptsResult: """Send a prompts/list request. Args: cursor: Simple cursor string for pagination (deprecated, use params instead) params: Full pagination parameters including cursor and any future fields """ if params is not None and cursor is not None: raise ValueError("Cannot specify both cursor and params") if params is not None: request_params = params elif cursor is not None: request_params = types.PaginatedRequestParams(cursor=cursor) else: request_params = None return await self.send_request( types.ClientRequest(types.ListPromptsRequest(params=request_params)), types.ListPromptsResult, ) async def get_prompt(self, name: str, arguments: dict[str, str] | None = None) -> types.GetPromptResult: """Send a prompts/get request.""" return await self.send_request( types.ClientRequest( types.GetPromptRequest( params=types.GetPromptRequestParams(name=name, arguments=arguments), ) ), types.GetPromptResult, ) async def complete( self, ref: types.ResourceTemplateReference | types.PromptReference, argument: dict[str, str], context_arguments: dict[str, str] | None = None, ) -> types.CompleteResult: """Send a completion/complete request.""" context = None if context_arguments is not None: context = types.CompletionContext(arguments=context_arguments) return await self.send_request( types.ClientRequest( types.CompleteRequest( params=types.CompleteRequestParams( ref=ref, argument=types.CompletionArgument(**argument), context=context, ), ) ), types.CompleteResult, ) @overload @deprecated("Use list_tools(params=PaginatedRequestParams(...)) instead") async def list_tools(self, cursor: str | None) -> types.ListToolsResult: ... @overload async def list_tools(self, *, params: types.PaginatedRequestParams | None) -> types.ListToolsResult: ... @overload async def list_tools(self) -> types.ListToolsResult: ... async def list_tools( self, cursor: str | None = None, *, params: types.PaginatedRequestParams | None = None, ) -> types.ListToolsResult: """Send a tools/list request. Args: cursor: Simple cursor string for pagination (deprecated, use params instead) params: Full pagination parameters including cursor and any future fields """ if params is not None and cursor is not None: raise ValueError("Cannot specify both cursor and params") if params is not None: request_params = params elif cursor is not None: request_params = types.PaginatedRequestParams(cursor=cursor) else: request_params = None result = await self.send_request( types.ClientRequest(types.ListToolsRequest(params=request_params)), types.ListToolsResult, ) # Cache tool output schemas for future validation # Note: don't clear the cache, as we may be using a cursor for tool in result.tools: self._tool_output_schemas[tool.name] = tool.outputSchema return result async def send_roots_list_changed(self) -> None: # pragma: no cover """Send a roots/list_changed notification.""" await self.send_notification(types.ClientNotification(types.RootsListChangedNotification())) async def _received_request(self, responder: RequestResponder[types.ServerRequest, types.ClientResult]) -> None: ctx = RequestContext[ClientSession, Any]( request_id=responder.request_id, meta=responder.request_meta, session=self, lifespan_context=None, ) # Delegate to experimental task handler if applicable if self._task_handlers.handles_request(responder.request): with responder: await self._task_handlers.handle_request(ctx, responder) return None # Core request handling match responder.request.root: case types.CreateMessageRequest(params=params): with responder: # Check if this is a task-augmented request if params.task is not None: response = await self._task_handlers.augmented_sampling(ctx, params, params.task) else: response = await self._sampling_callback(ctx, params) client_response = ClientResponse.validate_python(response) await responder.respond(client_response) case types.ElicitRequest(params=params): with responder: # Check if this is a task-augmented request if params.task is not None: response = await self._task_handlers.augmented_elicitation(ctx, params, params.task) else: response = await self._elicitation_callback(ctx, params) client_response = ClientResponse.validate_python(response) await responder.respond(client_response) case types.ListRootsRequest(): with responder: response = await self._list_roots_callback(ctx) client_response = ClientResponse.validate_python(response) await responder.respond(client_response) case types.PingRequest(): # pragma: no cover with responder: return await responder.respond(types.ClientResult(root=types.EmptyResult())) case _: # pragma: no cover pass # Task requests handled above by _task_handlers return None async def _handle_incoming( self, req: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception, ) -> None: """Handle incoming messages by forwarding to the message handler.""" await self._message_handler(req) async def _received_notification(self, notification: types.ServerNotification) -> None: """Handle notifications from the server.""" # Process specific notification types match notification.root: case types.LoggingMessageNotification(params=params): await self._logging_callback(params) case types.ElicitCompleteNotification(params=params): # Handle elicitation completion notification # Clients MAY use this to retry requests or update UI # The notification contains the elicitationId of the completed elicitation pass case _: pass

class ClientSessionGroup: """Client for managing connections to multiple MCP servers. This class is responsible for encapsulating management of server connections. It aggregates tools, resources, and prompts from all connected servers. For auxiliary handlers, such as resource subscription, this is delegated to the client and can be accessed via the session. Example Usage: name_fn = lambda name, server_info: f"{(server_info.name)}_{name}" async with ClientSessionGroup(component_name_hook=name_fn) as group: for server_param in server_params: await group.connect_to_server(server_param) ... """ class _ComponentNames(BaseModel): """Used for reverse index to find components.""" prompts: set[str] = set() resources: set[str] = set() tools: set[str] = set() # Standard MCP components. _prompts: dict[str, types.Prompt] _resources: dict[str, types.Resource] _tools: dict[str, types.Tool] # Client-server connection management. _sessions: dict[mcp.ClientSession, _ComponentNames] _tool_to_session: dict[str, mcp.ClientSession] _exit_stack: contextlib.AsyncExitStack _session_exit_stacks: dict[mcp.ClientSession, contextlib.AsyncExitStack] # Optional fn consuming (component_name, serverInfo) for custom names. # This is provide a means to mitigate naming conflicts across servers. # Example: (tool_name, serverInfo) => "{result.serverInfo.name}.{tool_name}" _ComponentNameHook: TypeAlias = Callable[[str, types.Implementation], str] _component_name_hook: _ComponentNameHook | None def __init__( self, exit_stack: contextlib.AsyncExitStack | None = None, component_name_hook: _ComponentNameHook | None = None, ) -> None: """Initializes the MCP client.""" self._tools = {} self._resources = {} self._prompts = {} self._sessions = {} self._tool_to_session = {} if exit_stack is None: self._exit_stack = contextlib.AsyncExitStack() self._owns_exit_stack = True else: self._exit_stack = exit_stack self._owns_exit_stack = False self._session_exit_stacks = {} self._component_name_hook = component_name_hook async def __aenter__(self) -> Self: # pragma: no cover # Enter the exit stack only if we created it ourselves if self._owns_exit_stack: await self._exit_stack.__aenter__() return self async def __aexit__( self, _exc_type: type[BaseException] | None, _exc_val: BaseException | None, _exc_tb: TracebackType | None, ) -> bool | None: # pragma: no cover """Closes session exit stacks and main exit stack upon completion.""" # Only close the main exit stack if we created it if self._owns_exit_stack: await self._exit_stack.aclose() # Concurrently close session stacks. async with anyio.create_task_group() as tg: for exit_stack in self._session_exit_stacks.values(): tg.start_soon(exit_stack.aclose) @property def sessions(self) -> list[mcp.ClientSession]: """Returns the list of sessions being managed.""" return list(self._sessions.keys()) # pragma: no cover @property def prompts(self) -> dict[str, types.Prompt]: """Returns the prompts as a dictionary of names to prompts.""" return self._prompts @property def resources(self) -> dict[str, types.Resource]: """Returns the resources as a dictionary of names to resources.""" return self._resources @property def tools(self) -> dict[str, types.Tool]: """Returns the tools as a dictionary of names to tools.""" return self._tools @overload async def call_tool( self, name: str, arguments: dict[str, Any], read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, *, meta: dict[str, Any] | None = None, ) -> types.CallToolResult: ... @overload @deprecated("The 'args' parameter is deprecated. Use 'arguments' instead.") async def call_tool( self, name: str, *, args: dict[str, Any], read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, meta: dict[str, Any] | None = None, ) -> types.CallToolResult: ... async def call_tool( self, name: str, arguments: dict[str, Any] | None = None, read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, *, meta: dict[str, Any] | None = None, args: dict[str, Any] | None = None, ) -> types.CallToolResult: """Executes a tool given its name and arguments.""" session = self._tool_to_session[name] session_tool_name = self.tools[name].name return await session.call_tool( session_tool_name, arguments if args is None else args, read_timeout_seconds=read_timeout_seconds, progress_callback=progress_callback, meta=meta, ) async def disconnect_from_server(self, session: mcp.ClientSession) -> None: """Disconnects from a single MCP server.""" session_known_for_components = session in self._sessions session_known_for_stack = session in self._session_exit_stacks if not session_known_for_components and not session_known_for_stack: raise McpError( types.ErrorData( code=types.INVALID_PARAMS, message="Provided session is not managed or already disconnected.", ) ) if session_known_for_components: # pragma: no cover component_names = self._sessions.pop(session) # Pop from _sessions tracking # Remove prompts associated with the session. for name in component_names.prompts: if name in self._prompts: del self._prompts[name] # Remove resources associated with the session. for name in component_names.resources: if name in self._resources: del self._resources[name] # Remove tools associated with the session. for name in component_names.tools: if name in self._tools: del self._tools[name] if name in self._tool_to_session: del self._tool_to_session[name] # Clean up the session's resources via its dedicated exit stack if session_known_for_stack: session_stack_to_close = self._session_exit_stacks.pop(session) # pragma: no cover await session_stack_to_close.aclose() # pragma: no cover async def connect_with_session( self, server_info: types.Implementation, session: mcp.ClientSession ) -> mcp.ClientSession: """Connects to a single MCP server.""" await self._aggregate_components(server_info, session) return session async def connect_to_server( self, server_params: ServerParameters, session_params: ClientSessionParameters | None = None, ) -> mcp.ClientSession: """Connects to a single MCP server.""" server_info, session = await self._establish_session(server_params, session_params or ClientSessionParameters()) return await self.connect_with_session(server_info, session) async def _establish_session( self, server_params: ServerParameters, session_params: ClientSessionParameters, ) -> tuple[types.Implementation, mcp.ClientSession]: """Establish a client session to an MCP server.""" session_stack = contextlib.AsyncExitStack() try: # Create read and write streams that facilitate io with the server. if isinstance(server_params, StdioServerParameters): client = mcp.stdio_client(server_params) read, write = await session_stack.enter_async_context(client) elif isinstance(server_params, SseServerParameters): client = sse_client( url=server_params.url, headers=server_params.headers, timeout=server_params.timeout, sse_read_timeout=server_params.sse_read_timeout, ) read, write = await session_stack.enter_async_context(client) else: httpx_client = create_mcp_http_client( headers=server_params.headers, timeout=httpx.Timeout( server_params.timeout.total_seconds(), read=server_params.sse_read_timeout.total_seconds(), ), ) await session_stack.enter_async_context(httpx_client) client = streamable_http_client( url=server_params.url, http_client=httpx_client, terminate_on_close=server_params.terminate_on_close, ) read, write, _ = await session_stack.enter_async_context(client) session = await session_stack.enter_async_context( mcp.ClientSession( read, write, read_timeout_seconds=session_params.read_timeout_seconds, sampling_callback=session_params.sampling_callback, elicitation_callback=session_params.elicitation_callback, list_roots_callback=session_params.list_roots_callback, logging_callback=session_params.logging_callback, message_handler=session_params.message_handler, client_info=session_params.client_info, ) ) result = await session.initialize() # Session successfully initialized. # Store its stack and register the stack with the main group stack. self._session_exit_stacks[session] = session_stack # session_stack itself becomes a resource managed by the # main _exit_stack. await self._exit_stack.enter_async_context(session_stack) return result.serverInfo, session except Exception: # pragma: no cover # If anything during this setup fails, ensure the session-specific # stack is closed. await session_stack.aclose() raise async def _aggregate_components(self, server_info: types.Implementation, session: mcp.ClientSession) -> None: """Aggregates prompts, resources, and tools from a given session.""" # Create a reverse index so we can find all prompts, resources, and # tools belonging to this session. Used for removing components from # the session group via self.disconnect_from_server. component_names = self._ComponentNames() # Temporary components dicts. We do not want to modify the aggregate # lists in case of an intermediate failure. prompts_temp: dict[str, types.Prompt] = {} resources_temp: dict[str, types.Resource] = {} tools_temp: dict[str, types.Tool] = {} tool_to_session_temp: dict[str, mcp.ClientSession] = {} # Query the server for its prompts and aggregate to list. try: prompts = (await session.list_prompts()).prompts for prompt in prompts: name = self._component_name(prompt.name, server_info) prompts_temp[name] = prompt component_names.prompts.add(name) except McpError as err: # pragma: no cover logging.warning(f"Could not fetch prompts: {err}") # Query the server for its resources and aggregate to list. try: resources = (await session.list_resources()).resources for resource in resources: name = self._component_name(resource.name, server_info) resources_temp[name] = resource component_names.resources.add(name) except McpError as err: # pragma: no cover logging.warning(f"Could not fetch resources: {err}") # Query the server for its tools and aggregate to list. try: tools = (await session.list_tools()).tools for tool in tools: name = self._component_name(tool.name, server_info) tools_temp[name] = tool tool_to_session_temp[name] = session component_names.tools.add(name) except McpError as err: # pragma: no cover logging.warning(f"Could not fetch tools: {err}") # Clean up exit stack for session if we couldn't retrieve anything # from the server. if not any((prompts_temp, resources_temp, tools_temp)): del self._session_exit_stacks[session] # pragma: no cover # Check for duplicates. matching_prompts = prompts_temp.keys() & self._prompts.keys() if matching_prompts: raise McpError( # pragma: no cover types.ErrorData( code=types.INVALID_PARAMS, message=f"{matching_prompts} already exist in group prompts.", ) ) matching_resources = resources_temp.keys() & self._resources.keys() if matching_resources: raise McpError( # pragma: no cover types.ErrorData( code=types.INVALID_PARAMS, message=f"{matching_resources} already exist in group resources.", ) ) matching_tools = tools_temp.keys() & self._tools.keys() if matching_tools: raise McpError( types.ErrorData( code=types.INVALID_PARAMS, message=f"{matching_tools} already exist in group tools.", ) ) # Aggregate components. self._sessions[session] = component_names self._prompts.update(prompts_temp) self._resources.update(resources_temp) self._tools.update(tools_temp) self._tool_to_session.update(tool_to_session_temp) def _component_name(self, name: str, server_info: types.Implementation) -> str: if self._component_name_hook: return self._component_name_hook(name, server_info) return name