You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
80 lines
3.5 KiB
80 lines
3.5 KiB
"""自定义工具执行器。
|
|
|
|
提供执行自定义 HTTP 工具的功能，支持多种认证方式（API Key、Bearer Token）和 HTTP 方法。
|
|
"""
|
|
import httpx
|
|
import json
|
|
|
|
|
|
class CustomToolExecutor:
|
|
"""自定义工具执行器类。
|
|
|
|
根据工具定义（端点 URL、HTTP 方法、认证配置等）执行 HTTP 请求。
|
|
|
|
Attributes:
|
|
endpoint_url: API 端点的基础 URL。
|
|
method: HTTP 请求方法。
|
|
path: API 路径。
|
|
headers: 请求头字典。
|
|
auth_type: 认证类型（none/api_key/bearer）。
|
|
auth_config: 认证配置信息。
|
|
timeout: 请求超时时间（秒）。
|
|
"""
|
|
|
|
def __init__(self, tool_def: dict):
|
|
"""初始化工具执行器。
|
|
|
|
Args:
|
|
tool_def: 工具定义字典，包含 endpoint_url、method、path、headers_json、auth_type、auth_config、timeout 等字段。
|
|
"""
|
|
self.endpoint_url = tool_def.get("endpoint_url", "") # API 基础 URL
|
|
self.method = tool_def.get("method", "GET") # HTTP 请求方法，默认为 GET
|
|
self.path = tool_def.get("path", "") # API 路径
|
|
self.headers = dict(tool_def.get("headers_json", {})) # 请求头字典
|
|
self.auth_type = tool_def.get("auth_type", "none") # 认证类型，默认为无认证
|
|
self.auth_config = dict(tool_def.get("auth_config", {})) # 认证配置信息
|
|
self.timeout = int(tool_def.get("timeout", 30)) # 请求超时时间（秒），默认 30 秒
|
|
|
|
async def execute(self, params: dict) -> str:
|
|
"""执行自定义工具请求。
|
|
|
|
根据工具定义构造完整的 URL，应用认证信息，发送 HTTP 请求并返回响应结果。
|
|
|
|
Args:
|
|
params: 请求参数，GET 请求作为查询参数，其他方法作为 JSON 请求体。
|
|
|
|
Returns:
|
|
str: 响应内容的字符串表示，最大长度 4000 字符。优先返回 JSON 格式，否则返回纯文本。
|
|
"""
|
|
# 构造完整 URL（确保基础 URL 和路径之间只有一个斜杠）
|
|
url = f"{self.endpoint_url.rstrip('/')}/{self.path.lstrip('/')}"
|
|
headers = dict(self.headers) # 复制请求头
|
|
req_params = dict(params) # 复制请求参数
|
|
|
|
# 根据认证类型添加认证信息
|
|
if self.auth_type == "api_key":
|
|
key = self.auth_config.get("key", "") # API Key
|
|
loc = self.auth_config.get("location", "header") # 认证位置（header/query）
|
|
name = self.auth_config.get("name", "X-API-Key") # 认证参数名
|
|
if loc == "header":
|
|
headers[name] = key # 添加到请求头
|
|
else:
|
|
req_params[name] = key # 添加到查询参数
|
|
elif self.auth_type == "bearer":
|
|
headers["Authorization"] = f"Bearer {self.auth_config.get('token', '')}" # Bearer Token 认证
|
|
|
|
timeout = httpx.Timeout(self.timeout) # 创建超时配置
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
if self.method == "GET":
|
|
resp = await client.get(url, params=req_params, headers=headers)
|
|
else:
|
|
resp = await client.request(
|
|
self.method, url, json=req_params, headers=headers
|
|
)
|
|
|
|
# 尝试解析 JSON 响应，否则返回纯文本
|
|
try:
|
|
data = resp.json()
|
|
return json.dumps(data, ensure_ascii=False, indent=2)[:4000] # 格式化 JSON，限制最大长度
|
|
except Exception:
|
|
return resp.text[:4000] # 返回纯文本响应，限制最大长度
|