类class agentscope.mcp

MCPClient

class MCPClient(BaseModel)

源码Source agentscope/mcp/_mcp_client.py:32–593

MCPClient 管理一个 MCP 服务连接，支持 STDIO 与 HTTP 传输。它把远端工具发现为 ToolBase，可一次性注册进工具包，并负责生命周期与超时控制。位于 agentscope.mcp 包。核心成员包括 model_post_init、set_runtime_headers、connect、close、list_raw_tools、list_tools。

This class provides a unified interface for MCP connections, handling both stateful (persistent) and stateless (ephemeral) connections.

Stateful: Requires explicit connect() and close(), maintains a client
Stateless: No connect() needed, creates a temporary client per call

Private attributes: - _client: The connected MCP SDK client for stateful connections - _lifecycle_task: Owns the stateful client context from enter to exit - _close_requested: Signals the lifecycle task to exit the client context - _is_connected: Connection state flag - _cached_tools: Cached list of tools - _http_client: The live HTTP client, while one is open - _static_headers: Its headers before any runtime override - _runtime_headers: See :meth:set_runtime_headers

示例Example

# Stateful connection (STDIO or HTTP)
client = MCPClient(
    name="file_system",
    is_stateful=True,
    mcp_config=StdioMCPConfig(
        command="mcp-server-filesystem"
    )
)
await client.connect()
tools = await client.list_tools()
await client.close()

# Stateless connection (HTTP only)
client = MCPClient(
    name="weather_search",
    is_stateful=False,
    mcp_config=HttpMCPConfig(
        url="https://api.weather.com/mcp"
    )
)
# No connect() needed
tools = await client.list_tools()

核心精讲Deep dive

MCPClient：接入 MCP 服务MCPClient: connect MCP servers

MCPClient 管理一个 MCP 服务连接，支持 STDIO 与 HTTP 传输。它把远端工具发现为 ToolBase，可一次性注册进 Toolkit，并负责生命周期与超时控制。

MCPClient manages a connection to an MCP server over STDIO or HTTP. It discovers remote tools as ToolBase objects, registers them into a Toolkit in one step, and owns lifecycle and timeout handling.

STDIO 连接必须是有状态的，需要显式 connect 与 close。STDIO connections are stateful and require explicit connect and close calls.
HTTP 连接可以是有状态或无状态的，适合远端服务。HTTP connections may be stateful or stateless, suiting remote services.
enable_tools、disable_tools 与 execution_timeout 控制暴露范围与执行预算。enable_tools, disable_tools and execution_timeout control exposure and execution budgets.

连接文件系统 MCP 服务Connect to the filesystem MCP server

import asyncio

from agentscope.mcp import MCPClient, StdioMCPConfig
from agentscope.tool import Toolkit


async def main() -> None:
    client = MCPClient(
        name="filesystem",
        is_stateful=True,
        mcp_config=StdioMCPConfig(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "."],
        ),
    )

    await client.connect()
    try:
        toolkit = Toolkit(mcps=[client])
        tools = await client.list_tools()
        print([tool.name for tool in tools])
        print(toolkit)  # 交给 Agent 使用
    finally:
        await client.close()


asyncio.run(main())

属性摘要Property Summary

属性Property	说明Description
`is_connected`	获取是已连接。 Whether the client is currently connected.

属性摘要Attribute Summary

属性Attribute	类型Type	说明Description	值Value
`name`	str	MCP 名称。	`Field(title='MCP Name', description='The MCP name.')`
`is_stateful`	bool	`is_stateful` 字段。	`Field(title='Stateful', description='Whether this is a stateful connection that requires explicit connect() and close(). STDIO MCP must be stateful. HTTP MCP can be either stateful or stateless.')`
`mcp_config`	StdioMCPConfig \| HttpMCPConfig	MCP 服务端配置。	`Field(discriminator='type', title='MCP Config', description='The MCP server configuration.')`
`enable_tools`	list[str] \| None	启用工具字段。	`None`
`disable_tools`	list[str] \| None	禁用工具字段。	`None`
`execution_timeout`	float \| None	执行超时字段。	`None`
`stream_progress`	bool	`stream_progress` 字段。	`Field(default=False, description='Whether local tools return a progress/result stream instead of the final ToolChunk. Sandbox gateway tools return final results only.')`

方法摘要Method Summary

方法Method	说明Description
`def model_post_init(__context: Any) -> None`	校验配置并初始化客户端。 Validate configuration and initialize client.
`async def set_runtime_headers(headers: dict[str, str]) -> None`async	为后续 MCP 请求设置运行时请求头。 Replace the headers sent with subsequent HTTP requests.
`async def connect() -> None`async	建立与 MCP 服务的连接。 Connect to the MCP server (for stateful connections only).
`async def close(ignore_errors: bool = True) -> None`async	关闭连接并释放底层资源。 Close the MCP connection (for stateful connections only).
`async def list_raw_tools() -> list[mcp.types.Tool]`async	列出原始工具。 List available tools from the MCP server in raw :class:`mcp.types.Tool` form, applying ``enable_tools`` and ``disable_tools`` filtering.
`async def list_tools() -> list[ToolBase]`async	把远端 MCP 工具发现为 ToolBase 列表。 List available tools from the MCP server as wrapped :class:`ToolBase` instances. If `enable_tools` and `disable_tools` are not `None` in the constructor, the returned tools will be filtered accordingly.
`async def get_tool(name: str) ->` `MCPTool`async	按名称获取单个 MCP 工具。 Get a tool by name from the MCP server.

属性详情Attribute Detail

name

str = Field(title='MCP Name', description='The MCP name.')

MCP 名称。

The MCP name.

源码Source agentscope/mcp/_mcp_client.py:92

is_stateful

bool = Field(title='Stateful', description='Whether this is a stateful connection that requires explicit connect() and close(). STDIO MCP must be stateful. HTTP MCP can be either stateful or stateless.')

is_stateful 字段。声明为 bool。

Whether this is a stateful connection that requires explicit connect() and close().

源码Source agentscope/mcp/_mcp_client.py:97

mcp_config

StdioMCPConfig | HttpMCPConfig = Field(discriminator='type', title='MCP Config', description='The MCP server configuration.')

MCP 服务端配置。

The MCP server configuration.

源码Source agentscope/mcp/_mcp_client.py:106

enable_tools

list[str] | None = None

启用工具字段。声明为 list[str] | None。

The enable tools field.

源码Source agentscope/mcp/_mcp_client.py:112

disable_tools

list[str] | None = None

禁用工具字段。声明为 list[str] | None。

The disable tools field.

源码Source agentscope/mcp/_mcp_client.py:117

execution_timeout

float | None = None

执行超时字段。声明为 float | None。

The execution timeout field.

源码Source agentscope/mcp/_mcp_client.py:121

stream_progress

bool = Field(default=False, description='Whether local tools return a progress/result stream instead of the final ToolChunk. Sandbox gateway tools return final results only.')

stream_progress 字段。声明为 bool。

Whether local tools return a progress/result stream instead of the final ToolChunk.

源码Source agentscope/mcp/_mcp_client.py:124

属性详情Property Detail

is_connected

def is_connected() -> bool

获取是已连接。

Whether the client is currently connected.

返回Returns bool

返回布尔值，表示判断结果。True if connected, False otherwise.

源码Source agentscope/mcp/_mcp_client.py:144–150

方法详情Method Detail

model_post_init

def model_post_init(__context: Any) -> None

校验配置并初始化客户端。

Validate configuration and initialize client.

参数Parameters

名称Name	类型Type	说明Description
`__context`	Any	上下文（类型 `Any`）。The context.

返回Returns None

无返回值。No value is returned.

源码Source agentscope/mcp/_mcp_client.py:152–196

set_runtime_headers

async def set_runtime_headers(headers: dict[str, str]) -> None

async

设置调用 MCP 服务时使用的请求头。

The map is replaced rather than merged: an empty map drops all runtime overrides, so the static headers from :attr:mcp_config apply again.

参数Parameters

名称Name	类型Type	说明Description
`headers`	dict[str, str]	请求头（类型 `dict[str, str]`）。The complete runtime header map. An empty dict clears it.

返回Returns None

无返回值。No value is returned.

异常Raises

可能抛出 `Exception`。`ValueError`:
可能抛出 `Exception`。The client is not Streamable HTTP, or a header is
可能抛出 `Exception`。invalid or owned by the HTTP layer.

源码Source agentscope/mcp/_mcp_client.py:280–335

connect

async def connect() -> None

async

建立连接并完成初始化。

For stateless connections, this method does nothing.

返回Returns None

无返回值。No value is returned.

异常Raises

RuntimeError 可能抛出 `RuntimeError`。If already connected.

源码Source agentscope/mcp/_mcp_client.py:337–392

close

async def close(ignore_errors: bool = True) -> None

async

关闭连接并释放相关资源。

For stateless connections, this method does nothing.

参数Parameters

名称Name	类型Type	说明Description
`ignore_errors`默认default `True`	bool	ignore 错误（类型 `bool`）。Whether to ignore errors during cleanup.

返回Returns None

无返回值。No value is returned.

异常Raises

RuntimeError 可能抛出 `RuntimeError`。If not connected.

源码Source agentscope/mcp/_mcp_client.py:413–460

list_raw_tools

async def list_raw_tools() -> list[mcp.types.Tool]

async

列出原始工具。

The full (unfiltered) tool list is cached on _cached_tools so :meth:get_tool can resolve names that were filtered out as well.

返回Returns list[mcp.types.Tool]

返回MCP.类型.工具列表。`list[mcp.types.Tool]`:

异常Raises

RuntimeError 可能抛出 `RuntimeError`。If not connected (for stateful connections).

源码Source agentscope/mcp/_mcp_client.py:466–503

list_tools

async def list_tools() -> list[ToolBase]

async

列出当前连接可用的工具。

List available tools from the MCP server as wrapped :class:ToolBase instances.

返回Returns list[ToolBase]

返回ToolBase列表。`list[ToolBase]`:

异常Raises

RuntimeError 可能抛出 `RuntimeError`。If not connected (for stateful connections).

源码Source agentscope/mcp/_mcp_client.py:505–519

get_tool

async def get_tool(name: str) -> MCPTool

async

获取工具通过名称来自 MCP 服务端。

The returned MCPTool object implements ToolProtocol and can be: - Called directly for its final result: await tool(arg1=val1) - Streamed when stream_progress=True: async for chunk in await tool(arg1=val1) - Registered to toolkit: toolkit.register_tool(tool)

参数Parameters

名称Name	类型Type	说明Description
`name`	str	工具函数到获取的名称。The name of the tool function to get.

返回Returns MCPTool

返回MCPTool。A tool object that implements ToolProtocol.

异常Raises

ValueError 当If 工具是不找到时抛出 `ValueError`。If the tool is not found.
RuntimeError 可能抛出 `RuntimeError`。If not connected (for stateful connections).

源码Source agentscope/mcp/_mcp_client.py:521–576