第 3 章：Tools 工具系统

3.1 本章目标

完成本章学习后，你将具备以下能力：

1. 掌握 DeepAgents 全部 10 个内置工具的名称、参数、返回值和使用场景
2. 掌握三种自定义工具的方式：@tool装饰器、StructuredTool、BaseTool
3. 能够通过MultiServerMCPClient连接外部 MCP Server 获取工具
4. 理解ToolCallLimitMiddleware和ToolRetryMiddleware的配置与使用
5. 能够使用permissions参数和FilesystemPermission控制工具权限

3.2 核心概念

3.2.1 工具系统的设计哲学

把 DeepAgents 的工具系统想象成一个瑞士军刀：

• 内置工具是军刀自带的基础功能（刀片、剪刀、开瓶器）-- 每个 Agent 都有
• 自定义工具是你可以加装的扩展模块 – 根据任务需要灵活添加
• MCP 工具是连接到外部服务的接口 – 像 USB 外设一样即插即用

3.2.2 内置工具完整列表

3.3 实战一：自定义工具三种方式

场景

创建一个能查询数据库、计算数学表达式、发送通知的工具集。

完整代码

# custom_tools_demo.pyfromtypingimportOptionalfrompydanticimportBaseModel,Fieldfromlangchain_core.toolsimporttool,StructuredTool,BaseToolfromdeepagentsimportcreate_deep_agent# ============================================================# 方式一：@tool 装饰器（推荐，最简单）# ============================================================@tooldefquery_database(sql:str)->str:"""Execute a SQL query against the database. Args: sql: The SQL query to execute (SELECT only for safety). Returns: The query results as a formatted string. """# 模拟数据库查询if"users"insql.lower():return"Query result:\n| id | name | email |\n| 1 | Alice | alice@test.com |\n| 2 | Bob | bob@test.com |"return"Query executed successfully. No results."@tooldefcalculate(expression:str)->str:"""Evaluate a mathematical expression safely. Args: expression: A mathematical expression (e.g., '2 + 3 * 4'). Returns: The computed result. """try:# 安全计算：仅允许数字和基本运算符allowed=set("0123456789+-*/().% ")ifnotall(cinallowedforcinexpression):return"Error: expression contains disallowed characters."result=eval(expression)returnf"Result:{result}"exceptExceptionase:returnf"Error:{str(e)}"# ============================================================# 方式二：StructuredTool（需要结构化输入时使用）# ============================================================classNotificationInput(BaseModel):"""Input schema for sending notifications."""recipient:str=Field(description="The recipient's email or phone number")message:str=Field(description="The notification message content")priority:Optional[str]=Field(default="normal",description="Priority level: 'low', 'normal', or 'high'")defsend_notification(recipient:str,message:str,priority:str="normal")->str:"""Send a notification to a recipient. Args: recipient: The recipient's email or phone number. message: The notification message content. priority: Priority level: 'low', 'normal', or 'high'. """return(f"Notification sent to{recipient}"f"[Priority:{priority}]:{message}")notification_tool=StructuredTool.from_function(func=send_notification,name="send_notification",description="Send a notification to a recipient via email or SMS.",args_schema=NotificationInput,)# ============================================================# 方式三：BaseTool（需要完全控制生命周期时使用）# ============================================================classFileStatsTool(BaseTool):"""A tool that computes statistics about file contents."""name:str="file_stats"description:str=("Compute statistics about a file's content ""(line count, word count, character count).")def_run(self,file_path:str)->str:"""Compute file statistics."""try:withopen(file_path,"r")asf:content=f.read()lines=content.count("\n")+1words=len(content.split())chars=len(content)return(f"File:{file_path}\n"f"Lines:{lines}\n"f"Words:{words}\n"f"Characters:{chars}")exceptFileNotFoundError:returnf"Error: File '{file_path}' not found."exceptExceptionase:returnf"Error reading file:{str(e)}"# ============================================================# 创建 Agent 并注册所有工具# ============================================================agent=create_deep_agent(model="openai:gpt-4o-mini",tools=[query_database,calculate,notification_tool,FileStatsTool(),],system_prompt=("You are a helpful assistant with database access, ""calculation, notification, and file analysis capabilities."),)# 运行result=agent.invoke({"messages":[{"role":"user","content":"Query all users from the database, ""then calculate 100 * (1 + 0.05)^3, and send a notification ""to admin@test.com about the results."}]})formsginresult["messages"]:ifhasattr(msg,"type")andmsg.type=="ai"andmsg.content:print(f"[AI]:{msg.content[:300]}")

运行结果

[AI]: Let me handle these tasks step by step. 1. First, let me query the database for all users. 2. Then calculate the compound interest. 3. Finally, send a notification. [TOOL - query_database]: Query result: | id | name | email | | 1 | Alice | alice@test.com | | 2 | Bob | bob@test.com | [TOOL - calculate]: Result: 115.76250000000002 [TOOL - send_notification]: Notification sent to admin@test.com [Priority: normal]: Query found 2 users. Calculated value: 115.76 [AI]: Here's a summary of the results: - Found 2 users in the database: Alice and Bob - 100 * (1 + 0.05)^3 = 115.76 - A notification has been sent to admin@test.com

逐段解析

方式一：@tool装饰器– 这是最推荐的方式。将 docstring 的第一行作为工具描述，Args:部分自动解析为参数 Schema。DeepAgents 会从函数签名推断参数类型和默认值。

方式二：StructuredTool– 当需要复杂参数验证或自定义 Schema 时使用。NotificationInput是一个 Pydantic 模型，定义了参数的完整结构。StructuredTool.from_function()将函数和 Schema 绑定。

方式三：BaseTool– 当需要完全控制工具的生命周期（如初始化外部连接、资源清理）时使用。继承BaseTool并实现_run方法。注意需要显式定义name和description。

3.4 实战二：MCP 协议集成

场景

通过 MCP（Model Context Protocol）协议连接外部服务，获取文件系统工具。

完整代码

# mcp_integration_demo.pyimportasynciofromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromdeepagentsimportcreate_deep_agentasyncdefmain():# 1. 配置 MCP 客户端，连接多个 MCP Serverclient=MultiServerMCPClient({"filesystem":{"transport":"stdio",# 通过标准输入输出通信"command":"npx","args":["-y","@modelcontextprotocol/server-filesystem","/workspace",],},# 也可以连接 HTTP 方式的 MCP Server# "web_search": {# "transport": "http",# "url": "http://localhost:8000/mcp",# },})# 2. 获取 MCP Server 提供的所有工具mcp_tools=awaitclient.get_tools()print(f"从 MCP Server 获取了{len(mcp_tools)}个工具:")fortinmcp_tools:print(f" -{t.name}:{t.description}")# 3. 创建 Agent 并传入 MCP 工具agent=create_deep_agent(model="openai:gpt-4o-mini",tools=mcp_tools,# MCP 工具与自定义工具可以混合使用system_prompt=("You are a helpful assistant with filesystem access ""via MCP tools."),)# 4. 异步运行 Agentresult=awaitagent.ainvoke({"messages":[{"role":"user","content":"List the files in /workspace and tell me what you see.",}]},config={"configurable":{"thread_id":"demo-1"}},)formsginresult["messages"]:ifhasattr(msg,"type")andmsg.type=="ai"andmsg.content:print(f"\n[AI]:{msg.content}")if__name__=="__main__":asyncio.run(main())

运行结果

从 MCP Server 获取了 4 个工具: - read_file: Read the complete contents of a file... - write_file: Create a new file or overwrite... - list_directory: List files and directories... - search_files: Search for files matching a pattern... [AI]: I can see the following files in /workspace: - hello_deep_agent.py - custom_middleware_demo.py - custom_tools_demo.py - mcp_integration_demo.py

逐段解析

MCP 客户端配置：MultiServerMCPClient接收一个字典，键为 Server 名称，值为连接配置。支持两种传输方式：

• stdio：通过标准输入输出与本地进程通信（如npx启动的 Node.js 进程）
• http：通过 HTTP 协议与远程 MCP Server 通信

工具获取：client.get_tools()异步获取所有 MCP Server 提供的工具列表，返回 LangChain 兼容的BaseTool实例列表。

Agent 集成：MCP 工具可以直接传入tools参数，与自定义工具和内置工具一起使用。

3.5 实战三：工具错误处理与重试

完整代码

# tool_retry_demo.pyfromdeepagentsimportcreate_deep_agentfromlangchain.agents.middlewareimport(ToolCallLimitMiddleware,ToolRetryMiddleware,)fromlangchain_core.toolsimporttoolimportrandom@tooldefunreliable_api(query:str)->str:"""Call an unreliable external API that sometimes fails. Args: query: The query to send to the API. Returns: The API response. """# 模拟 30% 的失败率ifrandom.random()<0.3:raiseConnectionError("API temporarily unavailable. Please retry.")returnf"API response for '{query}': Success!"agent=create_deep_agent(model="openai:gpt-4o-mini",tools=[unreliable_api],system_prompt="You are a helpful assistant with access to an external API.",middleware=[# 工具重试中间件：自动重试失败的工具调用（指数退避）ToolRetryMiddleware(max_retries=3,backoff_factor=2.0,# 退避因子：1s, 2s, 4smax_backoff=10.0,# 最大退避时间),# 工具调用次数限制：防止无限重试ToolCallLimitMiddleware(run_limit=10,# 单次运行最多 10 次工具调用exit_behavior="end",# 超限后优雅终止),],)result=agent.invoke({"messages":[{"role":"user","content":"Call the API with query 'hello world'"}]})formsginresult["messages"]:ifhasattr(msg,"type")andmsg.type=="ai"andmsg.content:print(f"[AI]:{msg.content}")

运行结果

[TOOL - unreliable_api]: Attempt 1 failed: ConnectionError [TOOL - unreliable_api]: Attempt 2: Success! [AI]: The API returned: "API response for 'hello world': Success!"

3.6 API 列表速查

工具方法速查：

3.7 常见错误与避坑

错误 1：工具函数参数类型不明确

# 错误：参数类型为 Any，DeepAgents 无法推断 Schema@tooldefmy_tool(data)->str:# data 类型不明确...# 正确：使用明确的类型注解@tooldefmy_tool(data:dict[str,str])->str:...

错误 2：MCP 连接未正确管理生命周期

# 错误：未正确关闭 MCP 客户端asyncdefmain():client=MultiServerMCPClient({...})tools=awaitclient.get_tools()# 忘记关闭 client# 正确：使用 context managerasyncdefmain():asyncwithMultiServerMCPClient({...})asclient:tools=awaitclient.get_tools()...

错误 3：工具重试配置不当导致无限循环

# 错误：max_retries 过大且没有 run_limitmiddleware=[ToolRetryMiddleware(max_retries=100),# 可能重试 100 次]# 正确：组合使用重试和限制middleware=[ToolRetryMiddleware(max_retries=3),ToolCallLimitMiddleware(run_limit=10),# 最多 10 次工具调用]

错误 4：BaseTool子类中_run和_arun不同步

# 错误：只实现了 _run，异步调用会失败classMyTool(BaseTool):def_run(self,query:str)->str:return"sync result"# 缺少 _arun# 正确：同时实现同步和异步版本classMyTool(BaseTool):def_run(self,query:str)->str:return"sync result"asyncdef_arun(self,query:str)->str:return"async result"

错误 5：permissions规则顺序错误

# 错误：宽泛的规则在前，具体的规则被遮蔽permissions=[FilesystemPermission(operations=["read"],paths=["/**"],mode="allow"),FilesystemPermission(operations=["read"],paths=["/workspace/.env"],mode="deny"),# /workspace/.env 的 deny 规则永远不会被匹配！]# 正确：具体的规则在前permissions=[FilesystemPermission(operations=["read"],paths=["/workspace/.env"],mode="deny"),FilesystemPermission(operations=["read"],paths=["/**"],mode="allow"),]

3.8 最佳实践

1. 工具函数 = 可预测的纯函数：避免副作用（side effects），相同输入始终产生相同输出，便于 LLM 理解和调试。
2. docstring 是工具的"用户手册"：LLM 根据 docstring 决定何时调用工具。写清楚功能、参数、返回值，必要时给出使用示例。
3. 优先使用@tool装饰器：它最简单，覆盖了 90% 的场景。仅在需要复杂验证时使用StructuredTool，仅在需要管理生命周期时使用BaseTool。
4. MCP 工具与自定义工具共存：DeepAgents 对工具来源透明，MCP 工具和自定义工具在 Agent 看来完全一样。
5. 始终配置工具调用次数限制：在生产环境中，ToolCallLimitMiddleware是防止失控的必备保险。

3.9 本章小结

1. DeepAgents 内置 10 个工具（ls、read_file、write_file、edit_file、delete、glob、grep、execute、task、write_todos），覆盖文件系统、子代理派发和任务规划三大领域。
2. 自定义工具推荐使用@tool装饰器，MCP 协议通过MultiServerMCPClient实现外部工具的无缝集成。
3. 生产环境中必须组合使用ToolRetryMiddleware（重试）和ToolCallLimitMiddleware（限流），确保 Agent 的稳定性和可控性。