Skip to main content
Client for managing MCP servers and sessions. This module provides a high-level client that manages MCP servers, connectors, and sessions from configuration.

MCPClient

from mcp_use.client.client import MCPClient

method init

Initialize a new MCP client.Parameters
config
str | dict[str, typing.Any] | None
default:"None"
Either a dict containing configuration or a path to a JSON config file.
allowed_servers
list[str] | None
default:"None"
Server name or configuration
sandbox
bool
default:"False"
Whether to use sandboxed execution mode for running MCP servers.
sandbox_options
mcp_use.client.connectors.sandbox.SandboxOptions | None
default:"None"
Optional sandbox configuration options.
sampling_callback
mcp.client.session.SamplingFnT | None
default:"None"
Optional sampling callback function.
elicitation_callback
mcp.client.session.ElicitationFnT | None
default:"None"
Callback function
message_handler
mcp.client.session.MessageHandlerFnT | None
default:"None"
Parameter value
logging_callback
mcp.client.session.LoggingFnT | None
default:"None"
Callback function
middleware
list[mcp_use.client.middleware.middleware.Middleware] | None
default:"None"
Middleware instance
code_mode
bool
default:"False"
Whether to enable code execution mode for tools.
verify
bool | None
default:"True"
Boolean flag
Signature
def __init__(config: str | dict[str, typing.Any] | None = None, allowed_servers: list[str] | None = None, sandbox: bool = False, sandbox_options: mcp_use.client.connectors.sandbox.SandboxOptions | None = None, sampling_callback: mcp.client.session.SamplingFnT | None = None, elicitation_callback: mcp.client.session.ElicitationFnT | None = None, message_handler: mcp.client.session.MessageHandlerFnT | None = None, logging_callback: mcp.client.session.LoggingFnT | None = None, middleware: list[mcp_use.client.middleware.middleware.Middleware] | None = None, code_mode: bool = False, verify: bool | None = True):

method add_middleware

Add a middleware.Parameters
middleware
mcp_use.client.middleware.middleware.Middleware
required
The middleware to add
Signature
def add_middleware(middleware: mcp_use.client.middleware.middleware.Middleware):

method add_server

Add a server configuration.Parameters
name
str
required
The name to identify this server.
server_config
dict[str, Any]
required
The server configuration.
Signature
def add_server(name: str, server_config: dict[str, Any]):

method close_all_sessions

Close all active sessions.This method ensures all sessions are closed even if some fail.Signature
def close_all_sessions():

method close_session

Close a session.Parameters
server_name
str
required
The name of the server to close the session for.
Signature
def close_session(server_name: str):

method create_all_sessions

Create sessions for all configured servers.Parameters
auto_initialize
bool
default:"True"
Whether to automatically initialize the sessions.
Returns
returns
dict[str, mcp_use.client.session.MCPSession]
Dictionary mapping server names to their MCPSession instances.
Signature
def create_all_sessions(auto_initialize: bool = True):

method create_session

Create a session for the specified server.Parameters
server_name
str
required
The name of the server to create a session for.
auto_initialize
bool
default:"True"
Whether to automatically initialize the session.
Returns
returns
mcp_use.client.session.MCPSession
The created MCPSession.
Signature
def create_session(server_name: str, auto_initialize: bool = True):

method execute_code

Execute Python code with access to MCP tools (code mode).This method allows agents to interact with MCP tools through Python code instead of direct tool calls, enabling more efficient context usage and data processing.Example:
client = MCPClient(config="config.json", code_mode=True)
await client.create_all_sessions()

result = await client.execute_code('''
tools = await search_tools("github")
print(f"Found \{len(tools)\} tools")

pr = await github.get_pull_request(
    owner="facebook",
    repo="react",
    number=12345
)
return \{"title": pr["title"]\}
''')

print(result['result'])  # \{'title': 'Fix bug in...'\}
print(result['logs'])    # ['Found 5 tools']
Parameters
code
str
required
Python code to execute with tool access.
timeout
float
default:"30.0"
Execution timeout in seconds.
Returns
returns
dict[str, Any]
Signature
def execute_code(code: str, timeout: float = 30.0):

method get_all_active_sessions

Get all active sessions.Returns
returns
dict[str, mcp_use.client.session.MCPSession]
Dictionary mapping server names to their MCPSession instances.
Signature
def get_all_active_sessions():

method get_server_names

Get the list of configured server names.Returns
returns
list[str]
List of server names (excludes internal code mode server).
Signature
def get_server_names():

method get_session

Get an existing session.Parameters
server_name
str
required
The name of the server to get the session for.
Returns
returns
mcp_use.client.session.MCPSession
The MCPSession for the specified server.
Signature
def get_session(server_name: str):

method remove_server

Remove a server configuration.Parameters
name
str
required
The name of the server to remove.
Signature
def remove_server(name: str):

method save_config

Save the current configuration to a file.Parameters
filepath
str
required
The path to save the configuration to.
Signature
def save_config(filepath: str):

method search_tools

Search available MCP tools across all active sessions.Example:
# Search for GitHub-related tools
result = await client.search_tools("github pull")
print(f"Found \{result['meta']['result_count']\} tools out of \{result['meta']['total_tools']\} total")
for tool in result['results']:
    print(f"\{tool['server']\}.\{tool['name']\}: \{tool['description']\}")
Parameters
query
str
default:""
Search query to filter tools by name or description.
detail_level
str
default:"full"
Level of detail to return:
Returns
returns
dict[str, Any]
Level of detail to return: - “names”: Only tool names and server - “descriptions”: Names, server, and descriptions - “full”: Complete tool information including schemas
Signature
def search_tools(query: str = "", detail_level: str = "full"):