3. 设计Agent的工具调用接口
任务目标
为Agent定义一个统一的工具调用接口，使Agent能够以一致的协议调用外部API或本地函数，并处理参数校验、执行与结果返回。本案例聚焦接口协议设计与最小可运行实现，不涉及具体业务工具的接入。
核心设计
工具接口采用声明式注册 + 动态分发模式。每个工具通过ToolSpec描述其名称、参数JSON Schema及执行函数；Agent通过ToolRegistry统一查找与调用。调用请求使用ToolCallRequest封装，返回ToolCallResult统一结构，包含执行状态、数据或错误信息。
完整实现
import json
from typing import Callable, Any, Dict, Optional
from dataclasses import dataclass, field
from enum import Enum
class ToolStatus(str, Enum):
SUCCESS = "success"
ERROR = "error"
NOT_FOUND = "not_found"
INVALID_ARGS = "invalid_args"
@dataclass
class ToolSpec:
name: str
description: str
parameters_schema: Dict[str, Any]
func: Callable[..., Any]
@dataclass
class ToolCallRequest:
tool_name: str
arguments: Dict[str, Any] = field(default_factory=dict)
@dataclass
class ToolCallResult:
status: ToolStatus
data: Optional[Any] = None
error: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return {
"status": self.status.value,
"data": self.data,
"error": self.error,
}
class ToolRegistry:
def __init__(self) -> None:
self._tools: Dict[str, ToolSpec] = {}
def register(self, spec: ToolSpec) -> None:
if spec.name in self._tools:
raise ValueError(f"Tool '{spec.name}' already registered")
self._tools[spec.name] = spec
def list_tools(self) -> list[Dict[str, Any]]:
return [
{
"name": spec.name,
"description": spec.description,
"parameters_schema": spec.parameters_schema,
}
for spec in self._tools.values()
]
def _validate_args(self, spec: ToolSpec, args: Dict[str, Any]) -> Optional[str]:
schema = spec.parameters_schema
required = schema.get("required", [])
properties = schema.get("properties", {})
for field_name in required:
if field_name not in args:
return f"Missing required argument: '{field_name}'"
for key, value in args.items():
if key not in properties:
return f"Unknown argument: '{key}'"
expected_type = properties[key].get("type")
if expected_type == "string" and not isinstance(value, str):
return f"Argument '{key}' must be string"
if expected_type == "number" and not isinstance(value, (int, float)):
return f"Argument '{key}' must be number"
if expected_type == "boolean" and not isinstance(value, bool):
return f"Argument '{key}' must be boolean"
return None
def call(self, request: ToolCallRequest) -> ToolCallResult:
spec = self._tools.get(request.tool_name)
if spec is None:
return ToolCallResult(status=ToolStatus.NOT_FOUND, error=f"Tool '{request.tool_name}' not found")
validation_error = self._validate_args(spec, request.arguments)
if validation_error:
return ToolCallResult(status=ToolStatus.INVALID_ARGS, error=validation_error)
try:
result = spec.func(**request.arguments)
return ToolCallResult(status=ToolStatus.SUCCESS, data=result)
except Exception as e:
return ToolCallResult(status=ToolStatus.ERROR, error=str(e))
# ---------- 示例工具与调用 ----------
def get_weather(city: str, unit: str = "celsius") -> str:
"""模拟天气查询（仅演示接口协议）"""
return f"Weather in {city}: 22°{unit[0].upper()}"
def add_numbers(a: float, b: float) -> float:
return a + b
def build_demo_registry() -> ToolRegistry:
registry = ToolRegistry()
registry.register(ToolSpec(
name="get_weather",
description="Get current weather for a city",
parameters_schema={
"type": "object",
"properties": {
"city": {"type": "string"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["city"],
},
func=get_weather,
))
registry.register(ToolSpec(
name="add_numbers",
description="Add two numbers",
parameters_schema={
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"},
},
"required": ["a", "b"],
},
func=add_numbers,
))
return registry
# ---------- 最小调用入口 ----------
if __name__ == "__main__":
registry = build_demo_registry()
# 成功调用
req1 = ToolCallRequest(tool_name="get_weather", arguments={"city": "Beijing"})
print(json.dumps(registry.call(req1).to_dict(), ensure_ascii=False))
# 参数校验失败
req2 = ToolCallRequest(tool_name="get_weather", arguments={"city": 123})
print(json.dumps(registry.call(req2).to_dict(), ensure_ascii=False))
# 工具不存在
req3 = ToolCallRequest(tool_name="nonexistent_tool", arguments={})
print(json.dumps(registry.call(req3).to_dict(), ensure_ascii=False))
运行与验证
- 保存代码为
tool_interface.py，直接运行：python tool_interface.py - 预期输出（与代码逻辑严格对应）：
{"status": "success", "data": "Weather in Beijing: 22°C", "error": null}
{"status": "invalid_args", "data": null, "error": "Argument 'city' must be string"}
{"status": "not_found", "data": null, "error": "Tool 'nonexistent_tool' not found"}
验证通过条件：三条输出分别对应成功调用、参数类型错误、工具不存在三种分支，且status字段与预期一致。
接口扩展说明
实际接入外部API时，只需在ToolSpec.func中发起HTTP请求（如使用requests库），并将响应解析后返回。注册与调用协议无需改动。若需异步工具，可将func替换为async函数并增加异步调用入口，但核心接口协议保持一致。