只有当工具的执行逻辑已经不是“调用 GPA 算子”或“查询数据目录”这类内置模式时,才需要自定义业务处理。此方式会替代工具原本的业务执行阶段,由您自己负责产出结果。典型场景是把一个工具实现为对外部 REST API 的转发。
第一步:编写自己的业务逻辑
在产品包目录 /cfg/gpa_mcp/v2_handlers/ 下新建 weather_hooks.py:
import httpx
async def handle(ctx):
"""自定义 businessHandler:替代整个②段,自己负责产出 ctx.result。"""
city = ctx.arguments.get("city")
async with httpx.AsyncClient() as client:
resp = await client.get(f"https://example-weather-api/query", params={"city": city})
resp.raise_for_status()
data = resp.json()
ctx.result = {"success": True, "temperature": data["temp"], "city": city}
第二步:让工具使用这段逻辑
{
"name": "getWeather",
"businessHandler": "weather_hooks"
}
不写 gpaToolId 前缀推导时,显式配置的 businessHandler 字段优先生效(省略函数名时使用默认的 handle)。
写自定义 businessHandler 时要自己处理清楚失败情况——ctx.result 里没有 "success": true 时,流水线会认为业务失败,跳过后续的 postHandler。
使用了网络请求的示例应写成 async def 并使用 httpx/aiohttp 这类异步库,避免用 requests 等同步库阻塞事件循环,拖慢服务的整体并发处理能力。