实战工具:HTTP 服务类

真正的智能体十有八九要"上网":查天气、看时间、查汇率、搜百科,本质都是对某个 HTTP 接口发一次 GET。本章把外部服务封装成"工具":带超时地请求、把各种网络异常翻译成模型能读懂的一句话、用第 5 章的 @tool 装饰器登记进注册表参与第 6/8 章调用循环;断网时还能用本地 mock 服务继续联调。

统一封装:超时 + 异常翻译

工具函数第一条原则:绝不把裸异常丢给模型ConnectionError: [Errno -2] ... 这类信息模型看不懂,要统一成一句人话,模型才知道下一步怎么办。先写一个通用 GET 封装(依赖 pip install requests):

# agent_demo/http_tools.py —— 第 9 章
import os
import requests

def http_get_json(url, params=None, timeout=8):
    """GET 一个 JSON 接口;无论成败都返回对模型友好的字典。"""
    try:
        resp = requests.get(url, params=params, timeout=timeout)
        resp.raise_for_status()               # 非 2xx 状态码抛 HTTPError
        return {"ok": True, "data": resp.json()}
    except requests.exceptions.Timeout:
        return {"ok": False, "error": f"请求超时(超过 {timeout} 秒),请稍后重试"}
    except requests.exceptions.ConnectionError:
        return {"ok": False, "error": "无法连接服务器:请检查网络或接口地址"}
    except requests.exceptions.HTTPError:
        return {"ok": False, "error": f"服务返回错误状态码:{resp.status_code}"}
    except ValueError:
        return {"ok": False, "error": "服务返回的内容不是合法 JSON"}

要点:timeout 必设,防止接口挂起拖死整个智能体;返回值永远是 {"ok": True, "data": ...}{"ok": False, "error": "..."},调用方(无论手工还是第 6 章循环)都不必再 try/except。

配置:真实接口与本地 mock 一键切换

接口地址做成环境变量可覆盖,联调时把三个服务统一指到本地 mock:

MOCK = os.environ.get("HTTP_TOOL_MOCK") == "1"     # HTTP_TOOL_MOCK=1 走本地
if MOCK:                                            # 全指向本地 mock(见文末)
    WEATHER_API = TIME_API = RATE_API = "http://127.0.0.1:8000"
else:                                               # 默认公网免费接口,可覆盖
    WEATHER_API = os.environ.get("WEATHER_API", "https://wttr.in")
    TIME_API = os.environ.get("TIME_API", "https://timeapi.io")
    RATE_API = os.environ.get("RATE_API", "https://open.er-api.com")

天气工具:中文城市名要先 URL 编码

wttr.in 免费、无需 key、支持中文城市,作为示例接口;换成你本机能访问的任意公开 API 也可以,只需改地址与解析字段:

from urllib.parse import quote

def get_weather(city: str) -> str:
    """查询指定城市的当前天气(温度/天气现象/湿度)。"""
    r = http_get_json(f"{WEATHER_API}/{quote(city)}", {"format": "j1"})
    if not r["ok"]:
        return r["error"]
    cur = (r["data"].get("current_condition") or [{}])[0]
    desc = (cur.get("weatherDesc") or [{}])[0].get("value", "未知")
    return (f"{city} 当前天气:{desc},{cur.get('temp_C', '?')}℃,"
            f"体感 {cur.get('FeelsLikeC', '?')}℃,湿度 {cur.get('humidity', '?')}%")

中文与特殊字符放进 URL 路径前必须 urllib.parse.quote;而 params 交给 requests 时它自动编码,不需要手工处理。

时间与汇率工具

def get_current_time(zone: str = "Asia/Shanghai") -> str:
    """查询指定 IANA 时区的当前时间,如 Asia/Tokyo。"""
    r = http_get_json(f"{TIME_API}/api/time/current/zone", {"timeZone": zone})
    if not r["ok"]:
        return r["error"]
    d = r["data"]
    return f"{zone} 现在是 {d['date']} {d['time']}(星期 {d['dayOfWeek']},UTC{d['utcOffset']})"

def get_exchange_rate(base: str = "USD", target: str = "CNY") -> str:
    """汇率查询:1 单位 base 货币 = ? target 货币。"""
    r = http_get_json(f"{RATE_API}/v6/latest/{base.upper()}")
    if not r["ok"]:
        return r["error"]
    rate = r["data"]["rates"].get(target.upper())
    if rate is None:
        return f"暂无 {target.upper()} 汇率数据"
    return f"1 {base.upper()} = {rate:.4f} {target.upper()}"

if __name__ == "__main__":                    # 手工直测,不依赖模型
    print(get_weather("北京"))
    # 输出(示例,视接口返回而定):北京 当前天气:Light rain,12℃,体感 10℃,湿度 86%
    print(get_exchange_rate("USD", "CNY"))
    # 输出(示例,视接口返回而定):1 USD = 7.2834 CNY

接入第 5 章注册表:@tool 装饰器

第 5 章的 @tool 装饰器会自动完成三件事:把函数登记进 TOOL_REGISTRY、按函数签名生成 OpenAI 格式的 Schema 追加进 TOOL_SCHEMAScall_tool(name, args) 按名字统一分发。把上文的裸函数"包"进装饰器即可(第 5 章的天气工具本来就是留给你换真实接口的占位,同名覆盖即可):

# tools.py 中新增/替换的部分:把 http_tools.py 里的真实实现接入注册表
from http_tools import (get_weather as _real_weather,
                        get_current_time as _real_time,
                        get_exchange_rate as _real_rate)
from tools import tool     # 沿用第 5 章装饰器

@tool(description="查询指定城市的当前天气(温度/天气现象/湿度)",
      params={"city": "城市名,如 北京"})
def get_weather(city: str) -> str:
    return _real_weather(city)

@tool(description="查询指定 IANA 时区的当前时间",
      params={"zone": "IANA 时区,如 Asia/Tokyo"})
def get_current_time(zone: str = "Asia/Shanghai") -> str:
    return _real_time(zone)

@tool(description="查询两种货币的汇率",
      params={"base": "基准货币代码,如 USD", "target": "目标货币,如 CNY"})
def get_exchange_rate(base: str = "USD", target: str = "CNY") -> str:
    return _real_rate(base, target)

装饰器不改变函数本身:既能被 call_tool 分发,也能直接 get_weather("北京") 单测。更彻底的做法是直接改写 tools.py 里原模拟函数的函数体(装饰器保持不动),避免 TOOL_SCHEMAS 里出现两条同名记录;上面的二次注册写法适合快速验证,分发时以最后一次注册为准。验证注册与分发:

from tools import TOOL_SCHEMAS, call_tool

print([s["function"]["name"] for s in TOOL_SCHEMAS])
# 输出(示例):['add', 'get_current_time', 'get_weather', 'get_exchange_rate', ...]
print(call_tool("get_weather", {"city": "北京"}))
# 输出(示例,视接口返回而定):北京 当前天气:Light rain,12℃,体感 10℃,湿度 86%

注册之后任何循环代码都不用改:无论第 6 章 run_with_tools 还是第 8 章 run_agent,请求时都自动携带 TOOL_SCHEMAS,模型要求工具时循环里执行 call_tool(name, args),把这种"人话结果"以 tool 消息回填。此前问"北京天气"得到的是第 5 章写死的模拟数据,现在自动换成真实接口。

离线备选:本地 mock 服务

公网连不上时用标准库起一个"假接口"继续开发。mock 刻意复刻真实服务的少量返回字段,工具代码完全不用改:

# agent_demo/mock_http_api.py —— 运行:python mock_http_api.py
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs, unquote

WEATHER = {"北京": ("晴", "16", "15", "40"), "上海": ("阴", "14", "13", "68")}

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        u, q = urlparse(self.path), parse_qs(urlparse(self.path).query)
        if u.path.startswith("/api/time/current/zone"):            # 时间接口
            body = {"date": "2026-01-08", "time": "10:00:00",
                    "dayOfWeek": "Thursday", "utcOffset": "+08:00"}
        elif u.path.startswith("/v6/latest/"):                     # 汇率接口
            body = {"rates": {"CNY": 7.28, "JPY": 148.50}}
        elif q.get("format") == ["j1"]:                            # 天气:/{城市}?format=j1
            desc, temp, feels, humi = WEATHER.get(unquote(u.path)[1:],
                                                  ("晴", "20", "19", "50"))
            body = {"current_condition": [{"temp_C": temp, "FeelsLikeC": feels,
                                           "humidity": humi,
                                           "weatherDesc": [{"value": desc}]}]}
        else:                                                      # 其余一律 404
            self.send_response(404)
            self.send_header("Content-Length", "0"); self.end_headers(); return
        data = json.dumps(body, ensure_ascii=False).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers(); self.wfile.write(data)

if __name__ == "__main__":
    print("mock API 已启动:http://127.0.0.1:8000")
    HTTPServer(("127.0.0.1", 8000), Handler).serve_forever()

断网联调,两个终端各跑一条命令:

# 终端 1:先起 mock
python agent_demo/mock_http_api.py
# 终端 2:所有工具自动指向 http://127.0.0.1:8000
HTTP_TOOL_MOCK=1 python agent_demo/http_tools.py
# 输出(http_tools.py 自测打印):北京 当前天气:晴,16℃,体感 15℃,湿度 40%
# 输出(继续打印):1 USD = 7.2800 CNY

顺带验证异常翻译——本地必现、结果确定:

import http_tools as h
print(h.http_get_json("http://127.0.0.1:8000/no-such-api", timeout=2))   # mock 返回 404
# 输出:{'ok': False, 'error': '服务返回错误状态码:404'}
print(h.http_get_json("http://127.0.0.1:1/x", timeout=2))                # 连不上的端口
# 输出:{'ok': False, 'error': '无法连接服务器:请检查网络或接口地址'}

小结:把 HTTP 服务变成 Agent 工具只需三步——封装带 timeout 的 GET 并把异常翻译成人话、按接口文档解析字段、用第 5 章 @tool 装饰器登记进注册表;配合 HTTP_TOOL_MOCK=1 指向本地 mock,断网也能完整联调工具层。

笔记加载中…