Skip to main content
View the source code for this module on GitHub: https://github.com/mcp-use/mcp-use/blob/main/libraries/python/mcp_use/client/task_managers/base.py
Connection management for MCP implementations. This module provides an abstract base class for different types of connection managers used in MCP connectors.

​
ConnectionManager

from mcp_use.client.task_managers.base import ConnectionManager

​
method init

Initialize a new connection manager.Signature
def __init__():

​
method get_streams

Get the current connection streams.Returns
​
returns
mcp_use.client.task_managers.base.T | None
The current connection (typically a tuple of read_stream, write_stream) or None if not connected.
Signature
def get_streams():

​
method start

Start the connection manager and establish a connection.Returns
​
returns
mcp_use.client.task_managers.base.T
The established connection.
Signature
def start():

​
method stop

Stop the connection manager and close the connection.This method ensures graceful shutdown by waiting for the connection task to complete and cleanup to finish. If operations exceed the timeout, forced cleanup is performed to prevent resource leaks.Note: This method does not raise exceptions and guarantees graceful shutdown even if cleanup times out.Parameters
​
timeout
float | None
default:"30.0"
Maximum time to wait for cleanup in seconds (default: 30.0).
Signature
def stop(timeout: float | None = 30.0):