MCP 工具调用

Workflow 脚本可以调用在项目「MCP 集成」中配置的任意外部 MCP 服务器。

mcp(server_name) → _MCPClient

获取指定服务器的 MCP 客户端。

exa = mcp("exa")
result = exa.call("web_search_exa", {"query": "Python 3.13 新特性"})
print(result["content"][0]["text"])

raises:

• KeyError — server_name 未在项目 MCP 配置中找到
• ValueError — 服务器没有 URL（stdio 类型不支持）

---

_MCPClient.call(tool_name, arguments, timeout) → dict

同步调用一个 MCP 工具。

def call(
    self,
    tool_name: str,
    arguments: dict | None = None,
    timeout: int = 60,
) -> dict

参数	类型	默认值	说明
tool_name	str	必填	工具名称
arguments	dict | None	None	工具参数
timeout	int	60	超时秒数

返回值： 包含 content 列表的 dict，content 中每项有 type 和 text 字段。

result = mcp("exa").call("web_search_exa", {"query": "MCP 2025"})
for item in result["content"]:
    if item["type"] == "text":
        print(item["text"])

---

_MCPClient.list_tools() → list

列出服务器上所有可用工具。

for tool in mcp("context7").list_tools():
    print(tool["name"], "-", tool["description"])

---

_MCPClient.acall() / alist_tools()

异步版本，在 async def run 中配合 asyncio.gather 使用：

async def run(params: dict, reporter) -> None:
    import asyncio
    r1, r2 = await asyncio.gather(
        mcp("exa").acall("web_search_exa", {"query": "Python 3.13"}),
        mcp("context7").acall("resolve-library-id",
                              {"query": "fastapi", "libraryName": "fastapi"}),
    )
    print(r1["content"][0]["text"])
    print(r2["content"][0]["text"])

---

mcp_call() 简写

def mcp_call(server_name: str, tool_name: str, arguments: dict | None = None, timeout: int = 60) -> dict

等价于 mcp(server_name).call(tool_name, arguments, timeout=timeout)。

result = mcp_call("exa", "web_search_exa", {"query": "MCP protocol 2025"})

---

async_mcp_call() 简写

async def async_mcp_call(server_name: str, tool_name: str, arguments: dict | None = None, timeout: int = 60) -> dict

等价于 await mcp(server_name).acall(tool_name, arguments, timeout=timeout)。

async def run(params: dict, reporter) -> None:
    result = await async_mcp_call("exa", "web_search_exa", {"query": "..."})