🛠️ 自定义工具开发

自定义工具是 Agent 能力的扩展点。无论是对接企业内部 API、操作数据库，还是执行特定的业务逻辑，都需要遵循一套规范化的开发流程，确保工具的安全性、可靠性和可维护性。

🏗️ 开发模式

根据工具复杂度选择合适的开发模式：

模式	描述	适用场景	示例
函数装饰器	用装饰器标记函数，自动提取 Schema	简单函数、快速原型	LangChain @tool、Python 类型注解
类继承	继承 BaseTool 基类，实现 _run 方法	有状态工具、复杂逻辑	LangChain BaseTool、OpenAI function 封装
配置驱动	YAML/JSON 定义 Schema + 实现函数分离	需要非开发者定义工具	企业工具市场、低代码平台
MCP Server	实现 MCP 协议的标准化工具服务	跨平台复用、社区分享	MCP Python/TS SDK

示例：三种方式实现天气查询工具

方式一：函数装饰器（LangChain）

from langchain.tools import tool
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")
    unit: str = Field(default="celsius", description="温度单位")

@tool(args_schema=WeatherInput)
def get_weather(city: str, unit: str = "celsius") -> str:
    """获取指定城市的天气信息"""
    # 实际调用天气 API
    return f"{city}天气：晴，28{unit}"

# 自动生成 Schema，直接可用
print(get_weather.name)        # "get_weather"
print(get_weather.description) # "获取指定城市的天气信息"
print(get_weather.args)        # WeatherInput schema

方式二：MCP Server（Python SDK）

from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationCapabilities
from mcp.types import Tool, TextContent
import mcp.server.stdio

server = Server("weather-server")

@server.list_tools()
async def handle_list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"},
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, args: dict) -> list[TextContent]:
    if name == "get_weather":
        city = args["city"]
        unit = args.get("unit", "celsius")
        # 实际调用天气 API
        return [TextContent(type="text", text=f"{city}天气：晴，28{unit}")]
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with mcp.server.stdio.stdio_server() as (r, w):
        await server.run(r, w, InitializationCapabilities(...))

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

🛡️ 安全最佳实践

自定义工具是 Agent 的"手和脚"，也是最容易出安全问题的地方。

安全清单

常见安全漏洞及防护

漏洞类型	攻击示例	防护措施
命令注入	LLM 传入 city="北京; rm -rf /"	✅ 避免使用 os.system/subprocess；必须使用时做参数白名单校验
路径遍历	LLM 传入 file="../../../etc/passwd"	✅ 规范化路径后检查是否在白名单目录内
Prompt 注入	LLM 输出中包含恶意指令，通过工具参数二次注入	✅ 工具结果与 system prompt 隔离；使用独立的消息结构
信息泄露	工具返回结果包含数据库连接串、内部 IP	✅ 工具输出做脱敏过滤；日志中敏感字段打码
DoS 攻击	LLM 反复调用耗时工具或传入超大参数	✅ 设置超时、最大输入大小、调用频率限制

🧪 测试方法

测试金字塔

各层测试要点

测试层	测试内容	工具/方法
单元测试	Schema 校验、参数解析、边界值、异常处理、幂等性	pytest、unittest
集成测试	对接真实 API/数据库，验证请求响应格式、超时处理	pytest + fixtures、WireMock（API mock）
E2E 测试	LLM 在真实对话中正确选择并调用工具，验证输出质量	手动测试 + 自动化评测脚本
安全测试	输入注入、权限绕过、敏感信息泄露	手动渗透测试 + 自动化安全扫描
性能测试	并发调用压力、大数据量处理、内存泄漏	locust、k6

📋 完整开发流程