agentscope.mcp¶

The MCP module in AgentScope, that provides fine-grained control over the MCP servers.

class MCPToolFunction[source]¶

Bases: object

An MCP tool function class that can be called directly.

__init__(mcp_name, tool, wrap_tool_result, client_gen=None, session=None)[source]¶

Initialize the MCP function.

Parameters:

mcp_name (str)
tool (Tool)
wrap_tool_result (bool)
client_gen (Callable[[...], _AsyncGeneratorContextManager[Any]] | None)
session (ClientSession | None)

Return type:

None

name: str¶: The name of the tool function.

description: str¶: The description of the tool function.

json_schema: dict[str, Any]¶: JSON schema of the tool function

async __call__(**kwargs)[source]¶

Call the MCP tool function with the given arguments, and return the result.

Parameters:: kwargs (Any)
Return type:: CallToolResult | ToolResponse

class MCPClientBase[source]¶

Bases: object

Base class for MCP clients.

__init__(name)[source]¶

Initialize the MCP client with a name.

Parameters:: name (str) – The name to identify the MCP server, which should be unique across the MCP servers.
Return type:: None

abstract async get_callable_function(func_name, wrap_tool_result=True)[source]¶

Get a tool function by its name.

Parameters:

func_name (str)
wrap_tool_result (bool)

Return type:

Callable

class StatefulClientBase[source]¶

Bases: MCPClientBase, ABC

The base class for stateful MCP clients in AgentScope, which maintains the session state across multiple tool calls.

The developers should use connect() and close() methods to manage the client lifecycle.

__init__(name)[source]¶

Initialize the stateful MCP client.

Parameters:: name (str) – The name to identify the MCP server, which should be unique across the MCP servers.
Return type:: None

is_connected: bool¶: If connected to the MCP server

async connect()[source]¶

Connect to MCP server.

Return type:: None

async close()[source]¶

Clean up the MCP client resources. You must call this method when your application is done.

Return type:: None

async list_tools()[source]¶

Get all available tools from the server.

Returns:: A list of available MCP tools.
Return type:: mcp.types.ListToolsResult

async get_callable_function(func_name, wrap_tool_result=True)[source]¶

Get an async tool function from the MCP server by its name, so that you can call it directly, wrap it into your own function, or anyway you like.

Note

Currently, only the text, image, and audio results are supported in this function.

Parameters:

func_name (str) – The name of the tool function to get.
wrap_tool_result (bool) – Whether to wrap the tool result into agentscope’s ToolResponse object. If False, the raw result type mcp.types.CallToolResult will be returned.

Returns:

A callable async function that returns either mcp.types.CallToolResult or ToolResponse when called.

Return type:

MCPToolFunction

class StdIOStatefulClient[source]¶

Bases: StatefulClientBase

A client class that sets up and manage StdIO MCP server connections, and provides function-level fine-grained control over the MCP servers.

Tip

The stateful client is recommended for MCP servers that need to maintain session states, e.g. web browsers or other interactive MCP servers.

Note

The stateful client will maintain one session across multiple tool calls, until the client is closed by explicitly calling the close() method.

__init__(name, command, args=None, env=None, cwd=None, encoding='utf-8', encoding_error_handler='strict')[source]¶

Initialize the MCP server with std IO.

Parameters:

name (str) – The name to identify the MCP server, which should be unique across the MCP servers.
command (str) – The executable to run to start the server.
args (list[str] | None, optional) – Command line arguments to pass to the executable.
env (dict[str, str] | None, optional) – The environment to use when spawning the process.
cwd (str | None, optional) – The working directory to use when spawning the process.
encoding (str, optional) – The text encoding used when sending/receiving messages to the server. Defaults to “utf-8”.
(`Literal["strict" (encoding_error_handler) –

defaults to “strict”):
The text encoding error handler.
"ignore" –

defaults to “strict”):
The text encoding error handler.
"replace"]` –

defaults to “strict”):
The text encoding error handler.
encoding_error_handler (Literal['strict', 'ignore', 'replace'])

Return type:

None

:param :

defaults to “strict”):: The text encoding error handler.

class HttpStatelessClient[source]¶

Bases: MCPClientBase

The sse/streamable HTTP MCP client implementation in AgentScope.

Note

Note this client is stateless, meaning it won’t maintain the session state across multiple tool calls. Each tool call will start a new session and close it after the call is done.

stateful: bool = False¶: Whether the MCP server is stateful, meaning it will maintain the session state across multiple tool calls, or stateless, meaning it will start a new session for each tool call.

__init__(name, transport, url, headers=None, timeout=30, sse_read_timeout=300, **client_kwargs)[source]¶

Initialize the streamable HTTP MCP server.

Parameters:

name (str) – The name to identify the MCP server, which should be unique across the MCP servers.
transport (Literal[“streamable_http”, “sse”]) – The transport type of MCP server. Generally, the URL of sse transport should end with /sse, while the streamable HTTP URL ends with /mcp.
url (str) – The URL of the MCP server.
headers (dict[str, str] | None, optional) – Additional headers to include in the HTTP request.
timeout (float, optional) – The timeout for the HTTP request in seconds. Defaults to 30.
sse_read_timeout (float, optional) – The timeout for reading Server-Sent Events (SSE) in seconds. Defaults to 300 (5 minutes).
**client_kwargs (Any) – The additional keyword arguments to pass to the streamable HTTP client.

Return type:

None

get_client()[source]¶

The disposable MCP client object, which is a context manager.

Return type:: _AsyncGeneratorContextManager[Any]

async get_callable_function(func_name, wrap_tool_result=True)[source]¶

Get a tool function by its name.

Parameters:

func_name (str) – The name of the tool function.
wrap_tool_result (bool, defaults to True) – Whether to wrap the tool result into agentscope’s ToolResponse object. If False, the raw result type mcp.types.CallToolResult will be returned.

Returns:

An async tool function that returns either mcp.types.CallToolResult or ToolResponse when called.

Return type:

Callable[…, Awaitable[mcp.types.CallToolResult | ToolResponse]]

async list_tools()[source]¶

List all tools available on the MCP server.

Returns:: The result containing the list of tools.
Return type:: mcp.types.ListToolsResult

class HttpStatefulClient[source]¶

Bases: StatefulClientBase

The stateful sse/streamable HTTP MCP client implementation in AgentScope.

Tip

The stateful client is recommended for MCP servers that need to maintain session states, e.g. web browsers or other interactive MCP servers.

Note

The stateful client will maintain one session across multiple tool calls, until the client is closed by explicitly calling the close() method.

__init__(name, transport, url, headers=None, timeout=30, sse_read_timeout=300, **client_kwargs)[source]¶

Initialize the streamable HTTP MCP client.

Parameters:

name (str) – The name to identify the MCP server, which should be unique across the MCP servers.
transport (Literal[“streamable_http”, “sse”]) – The transport type of MCP server. Generally, the URL of sse transport should end with /sse, while the streamable HTTP URL ends with /mcp.
url (str) – The URL to the MCP server.
headers (dict[str, str] | None, optional) – Additional headers to include in the HTTP request.
timeout (float, optional) – The timeout for the HTTP request in seconds. Defaults to 30.
sse_read_timeout (float, optional) – The timeout for reading Server-Sent Events (SSE) in seconds. Defaults to 300 (5 minutes).
**client_kwargs (Any) – The additional keyword arguments to pass to the streamable HTTP client.

Return type:

None