工具是运行在服务器端的函数,Client 可以携带参数来调用。使用 MCPClient 时,基本流程如下:
## 列出工具
查看服务器上有哪些可用的工具:
from dedalus_mcp.client import open_connection
async with open_connection("http://127.0.0.1:8000/mcp") as client:
tools = await client.list_tools()
for tool in tools.tools:
print(f"{tool.name}: {tool.description}") # 打印 tool 名称和描述
每个 tool 包含:
| 字段 | 类型 | 描述 |
|---|
name | str | tool 标识符 |
description | str | 对 tool 的功能说明 |
inputSchema | dict | 参数的 JSON Schema 定义 |
for tool in tools.tools:
print(f"Name: {tool.name}")
print(f"Description: {tool.description}")
print(f"Parameters schema: {tool.inputSchema}")
执行工具并传入参数:
result = await client.call_tool("add", {"a": 5, "b": 3})
# 工具结果是内容块;大多数工具返回 TextContent。
print(result.content[0].text)
如果服务器返回 JSON-RPC 错误(这在 tool 抛出异常时很常见),call_tool(...) 会抛出 McpError:
from mcp.shared.exceptions import McpError
try:
result = await client.call_tool("divide", {"a": 10, "b": 0})
print(result.content[0].text)
except McpError as e:
print(f"工具调用失败:{e}")
import asyncio
from dedalus_mcp.client import MCPClient
from mcp.shared.exceptions import McpError
async def main():
client = await MCPClient.connect("http://127.0.0.1:8000/mcp")
try:
tools = await client.list_tools()
print("可用工具:")
for tool in tools.tools:
print(f" - {tool.name}")
add_result = await client.call_tool("add", {"a": 5, "b": 3})
print(f"5 + 3 = {add_result.content[0].text}")
mul_result = await client.call_tool("multiply", {"a": 4, "b": 7})
print(f"4 * 7 = {mul_result.content[0].text}")
except McpError as e:
print(f"工具调用失败:{e}")
finally:
await client.close()
asyncio.run(main())
open_connection(...) 是一个异步上下文管理器。这意味着你无需手动调用 await client.close()。当异步 with 代码块退出时,它会自动为你关闭底层连接。
from dedalus_mcp.client import open_connection
async with open_connection("http://127.0.0.1:8000/mcp") as client:
tools = await client.list_tools()
result = await client.call_tool("greet", {"name": "World"})
print(result.content[0].text)
Last modified on April 11, 2026