TL;DR

系列方向： 从零搭建 AI Test Agent，逐步实现大模型接入、工具调用、任务规划、异常恢复和真实测试环境接入，最终形成一个可运行的 AI 测试辅助项目。

本篇重点： 前面几篇主要通过本地模拟工具验证 Agent 的执行能力。这一篇把工具接到 HTTP 接口上，让 Agent 能够查询真实的接口响应，并根据响应数据执行测试断言。

一、为什么要让 Agent 接入 HTTP 接口？

前面我们已经完成了几个基础模块：

• 第 01 篇：搭建 Agent 基础架构。
• 第 02 篇：接入大模型，让 Agent 理解测试任务。
• 第 03 篇：通过 Tool Calling 调用测试工具。
• 第 04 篇：构建 Agent Loop，支持多步骤任务执行。
• 第 05 篇：增加任务规划和状态管理。
• 第 06 篇：增加失败重试和异常恢复。

不过，前面使用的订单查询工具主要是本地模拟数据。

例如：

ORDERS = {
    "ORDER1001": {
        "amount": 199.00,
        "status": "paid"
    }
}

这种方式适合验证 Agent 的工具调用逻辑，但它和实际测试工作还有一定距离。

在真实测试项目中，测试人员面对的通常是：

1. 测试环境已经部署了一套业务系统。
2. 系统提供 HTTP API。
3. 测试工具向接口发送请求。
4. 接收响应并检查状态码、响应字段和业务规则。
5. 记录执行结果，必要时进行重试。

因此，本篇要解决的问题很明确：

如何让 Agent 不再只调用本地模拟函数，而是通过 HTTP 接口访问测试环境？

本篇使用订单查询作为示例。为了让读者可以直接复现，我们先搭建一个本地 HTTP 测试服务，再让 Agent 调用这个服务。后续只需要替换服务地址，就可以接入实际测试环境。

二、整体实现方案

这次不改变前面 Agent 的核心思路，而是在原有工具体系中增加 HTTP 接口工具。

整体执行过程如下：

这里有一个设计细节需要注意：

HTTP 请求成功，不代表业务测试成功。

例如：

{
  "code": 200,
  "data": {
    "order_id": "ORDER1002",
    "amount": -20.00,
    "status": "unknown"
  }
}

接口返回 HTTP 200，说明请求正常完成，但订单金额为负数、订单状态为 unknown，显然不符合我们的业务预期。

所以需要把执行结果分成两个层次：

层次	检查内容	结果示例
接口调用	HTTP 请求是否正常完成	请求成功
业务断言	响应数据是否符合预期	测试失败

这个区分会贯穿本篇代码。

三、先搭建一个本地 HTTP 测试服务

为了避免依赖外部业务系统，本篇使用 Python 标准库搭建一个简单的订单 API。

项目结构：

ai-test-agent/
├── mock_server.py       # 本地 HTTP 测试服务
├── http_tools.py        # HTTP 请求和断言工具
├── agent.py             # Agent 和大模型交互
├── main.py              # 项目入口
└── requirements.txt     # 项目依赖

本篇没有使用第三方 Web 框架，服务端和 HTTP 客户端均使用 Python 标准库。

1. 编写 HTTP 服务

新建 mock_server.py：

from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import time
from urllib.parse import urlparse, unquote


ORDERS = {
    "ORDER1001": {
        "order_id": "ORDER1001",
        "amount": 199.00,
        "status": "paid",
        "user_id": "USER001"
    },
    "ORDER1002": {
        "order_id": "ORDER1002",
        "amount": -20.00,
        "status": "unknown",
        "user_id": "USER002"
    }
}


class OrderHandler(BaseHTTPRequestHandler):

    def send_json(self, status, data):
        body = json.dumps(
            data,
            ensure_ascii=False
        ).encode("utf-8")

        self.send_response(status)
        self.send_header(
            "Content-Type",
            "application/json; charset=utf-8"
        )
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = unquote(urlparse(self.path).path)

        if path.startswith("/api/orders/"):
            order_id = path.split("/")[-1]

            # 模拟服务端超时场景
            if order_id == "SLOW":
                time.sleep(4)

            # 模拟服务端 500 错误
            if order_id == "SERVER_ERROR":
                self.send_json(
                    500,
                    {"message": "Internal Server Error"}
                )
                return

            order = ORDERS.get(order_id)

            if order is None:
                self.send_json(
                    404,
                    {"message": "Order not found"}
                )
                return

            self.send_json(
                200,
                {
                    "code": 200,
                    "message": "success",
                    "data": order
                }
            )
            return

        self.send_json(
            404,
            {"message": "API not found"}
        )

    def log_message(self, format, *args):
        print("[HTTP]", format % args)


if __name__ == "__main__":
    server = HTTPServer(
        ("127.0.0.1", 8000),
        OrderHandler
    )

    print("Mock API Server: http://127.0.0.1:8000")
    server.serve_forever()

这里准备了四种测试情况：

订单 ID	返回情况	测试目的
ORDER1001	HTTP 200，正常订单	验证正常业务
ORDER1002	HTTP 200，异常业务数据	验证业务断言
NOT_FOUND	HTTP 404	验证资源不存在
SERVER_ERROR	HTTP 500	验证服务端异常
SLOW	延迟响应	验证请求超时

注意：SLOW 是为了演示超时处理而故意延迟响应的，不代表真实业务接口一定会这样设计。

2. 启动服务

执行：

python mock_server.py

预期启动输出：

Mock API Server: http://127.0.0.1:8000

打开另一个终端，执行：

curl http://127.0.0.1:8000/api/orders/ORDER1001

预期返回：

{
  "code": 200,
  "message": "success",
  "data": {
    "order_id": "ORDER1001",
    "amount": 199.0,
    "status": "paid",
    "user_id": "USER001"
  }
}

到这里，本地测试服务就准备好了。

四、把 HTTP 请求封装成 Agent 工具

接下来是本篇最关键的部分。

前面我们通过工具注册表调用本地 Python 函数，现在只需要把 HTTP 请求也封装成工具，就能沿用 Agent 原有的调用方式。

这里我不建议让大模型直接传入任意 URL。

原因很简单：如果模型可以自由指定访问地址，就可能访问本机管理端口、内网服务或其他不应该访问的地址。

因此，本篇采用固定 BASE_URL 的方式：

• 接口地址由 Python 配置。
• 模型只能提供订单 ID。
• HTTP 方法固定为 GET。
• 请求超时时间由代码控制。
• 模型不能自行指定目标主机。

1. 编写 HTTP 工具

新建 http_tools.py：

import json
import os
import time
import socket

from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen


BASE_URL = os.getenv(
    "TEST_API_BASE_URL",
    "http://127.0.0.1:8000"
).rstrip("/")

REQUEST_TIMEOUT = 2
MAX_RETRIES = 2


def is_timeout_error(error):
    if isinstance(error, (TimeoutError, socket.timeout)):
        return True

    if isinstance(error, URLError):
        return isinstance(
            error.reason,
            (TimeoutError, socket.timeout)
        )

    return False


def query_order_detail(order_id):
    """
    查询订单详情。
    BASE_URL 固定由程序配置，模型不能传入任意 URL。
    """

    if not isinstance(order_id, str) or not order_id.strip():
        return {
            "success": False,
            "retryable": False,
            "error": "order_id 不能为空"
        }

    safe_order_id = quote(order_id.strip(), safe="")
    url = f"{BASE_URL}/api/orders/{safe_order_id}"

    for attempt in range(MAX_RETRIES + 1):
        request = Request(
            url,
            method="GET",
            headers={
                "Accept": "application/json"
            }
        )

        try:
            with urlopen(
                request,
                timeout=REQUEST_TIMEOUT
            ) as response:

                status_code = response.status
                body = response.read().decode("utf-8")

            result = json.loads(body)

            return {
                "success": True,
                "retryable": False,
                "http_status": status_code,
                "data": result.get("data")
            }

        except HTTPError as error:
            retryable = error.code in (429, 500, 502, 503, 504)

            if retryable and attempt < MAX_RETRIES:
                time.sleep(0.5 * (2 ** attempt))
                continue

            return {
                "success": False,
                "retryable": retryable,
                "http_status": error.code,
                "error": str(error)
            }

        except (TimeoutError, socket.timeout, URLError) as error:
            retryable = is_timeout_error(error) or isinstance(
                error, URLError
            )

            if retryable and attempt < MAX_RETRIES:
                time.sleep(0.5 * (2 ** attempt))
                continue

            return {
                "success": False,
                "retryable": retryable,
                "error": str(error)
            }

        except (UnicodeDecodeError, json.JSONDecodeError) as error:
            return {
                "success": False,
                "retryable": False,
                "error": f"响应数据解析失败: {error}"
            }

    return {
        "success": False,
        "retryable": False,
        "error": "请求失败，已达到最大重试次数"
    }


def check_order_amount(order, expected_amount):
    """
    检查订单金额。
    """

    actual_amount = order.get("amount")

    passed = (
        isinstance(actual_amount, (int, float))
        and not isinstance(actual_amount, bool)
        and actual_amount == expected_amount
    )

    return {
        "success": True,
        "passed": passed,
        "actual": actual_amount,
        "expected": expected_amount,
        "message": (
            "订单金额符合预期"
            if passed else "订单金额不符合预期"
        )
    }


def check_order_status(order, expected_status):
    """
    检查订单状态。
    """

    actual_status = order.get("status")
    passed = actual_status == expected_status

    return {
        "success": True,
        "passed": passed,
        "actual": actual_status,
        "expected": expected_status,
        "message": (
            "订单状态符合预期"
            if passed else "订单状态不符合预期"
        )
    }


TOOLS = {
    "query_order_detail": query_order_detail,
    "check_order_amount": check_order_amount,
    "check_order_status": check_order_status
}


def execute_tool(name, arguments):
    tool = TOOLS.get(name)

    if tool is None:
        return {
            "success": False,
            "retryable": False,
            "error": f"未知工具: {name}"
        }

    try:
        return tool(**arguments)

    except (TypeError, AttributeError) as error:
        return {
            "success": False,
            "retryable": False,
            "error": f"工具参数或数据格式错误: {error}"
        }

这段代码主要做了三件事：

第一，封装 HTTP 请求。

query_order_detail() 负责访问测试服务，不再直接读取本地 ORDERS 字典。

第二，增加请求超时和重试。

请求超时设置为 2 秒，最多额外重试 2 次。对于 429、500、502、503、504 等可能暂时恢复的情况，进行有限重试。

这里需要结合业务调整：不是所有 5xx 都一定能通过重试恢复，也不是所有连接错误都适合自动重试。生产环境中应进一步区分 DNS 错误、连接拒绝、超时等异常。

第三，保留独立的业务断言工具。

金额和状态检查不再重新访问模拟数据，而是直接使用 HTTP 查询工具返回的订单对象。

这样 Agent 的执行链路就变成：

query_order_detail
        |
        v
获取接口响应中的 data
        |
        v
check_order_amount
        |
        v
check_order_status

还有一个容易忽略的问题：HTTP 404、500 和请求超时都属于接口调用失败，不应该伪装成业务断言失败；而金额不符合预期，则应该记录为业务测试失败。

五、让 Agent 调用 HTTP 工具

前面已经有了 HTTP 工具，下一步就是把它注册到大模型的 Tool Calling 流程中。

这里给出一个独立的 OpenAI 兼容 Chat Completions 示例。它沿用前面文章的思路：模型选择工具，Python 执行工具，再把执行结果交还模型。

1. 配置模型服务

本篇通过环境变量读取模型配置：

export LLM_API_KEY="你的API_KEY"
export LLM_BASE_URL="https://你的模型服务地址/v1"
export LLM_MODEL="你的模型名称"

LLM_BASE_URL 应填写模型服务商提供的兼容 API 地址。不同服务商的接口兼容程度可能不同，需要确认其支持 Chat Completions 和 Tool Calling。

2. 编写 Agent

新建 agent.py：

import json
import os

from urllib.request import Request, urlopen

from http_tools import execute_tool


LLM_API_KEY = os.getenv("LLM_API_KEY")
LLM_BASE_URL = os.getenv("LLM_BASE_URL", "").rstrip("/")
LLM_MODEL = os.getenv("LLM_MODEL")


TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "query_order_detail",
            "description": "根据订单ID查询订单详情",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {
                        "type": "string",
                        "description": "订单ID"
                    }
                },
                "required": ["order_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "check_order_amount",
            "description": "检查订单金额是否符合预期",
            "parameters": {
                "type": "object",
                "properties": {
                    "order": {
                        "type": "object",
                        "description": "订单查询接口返回的订单数据"
                    },
                    "expected_amount": {
                        "type": "number",
                        "description": "预期金额"
                    }
                },
                "required": ["order", "expected_amount"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "check_order_status",
            "description": "检查订单状态是否符合预期",
            "parameters": {
                "type": "object",
                "properties": {
                    "order": {
                        "type": "object",
                        "description": "订单查询接口返回的订单数据"
                    },
                    "expected_status": {
                        "type": "string",
                        "description": "预期订单状态"
                    }
                },
                "required": ["order", "expected_status"]
            }
        }
    }
]


SYSTEM_PROMPT = """
你是一个软件测试 Agent。

你可以使用工具完成订单接口测试。

执行要求：
1. 用户要求查询订单时，先调用 query_order_detail。
2. 只有查询工具 success=true 且返回了订单 data，才能执行后续业务断言。
3. 金额检查调用 check_order_amount。
4. 状态检查调用 check_order_status。
5. 断言工具必须使用查询接口返回的订单对象，不能自行编造数据。
6. success 表示工具执行是否成功，passed 表示业务断言是否通过。
7. 如果接口请求失败，不要继续执行依赖该订单数据的断言。
8. 最后总结接口调用情况、业务断言结果和失败原因。
"""


class AITestAgent:

    def __init__(self, max_rounds=8):
        if not LLM_API_KEY or not LLM_BASE_URL or not LLM_MODEL:
            raise RuntimeError(
                "请先配置 LLM_API_KEY、LLM_BASE_URL 和 LLM_MODEL"
            )

        self.max_rounds = max_rounds

    def call_llm(self, messages):
        url = f"{LLM_BASE_URL}/chat/completions"

        payload = {
            "model": LLM_MODEL,
            "messages": messages,
            "tools": TOOL_SCHEMAS,
            "tool_choice": "auto",
            "temperature": 0
        }

        request = Request(
            url,
            data=json.dumps(payload).encode("utf-8"),
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {LLM_API_KEY}"
            },
            method="POST"
        )

        with urlopen(request, timeout=60) as response:
            result = json.loads(
                response.read().decode("utf-8")
            )

        return result["choices"][0]["message"]

    def run(self, task):
        messages = [
            {
                "role": "system",
                "content": SYSTEM_PROMPT
            },
            {
                "role": "user",
                "content": task
            }
        ]

        for round_number in range(self.max_rounds):
            response = self.call_llm(messages)

            tool_calls = response.get("tool_calls")

            if not tool_calls:
                return response.get("content", "Agent 未返回最终结果")

            # 将模型的工具调用消息加入上下文
            messages.append(response)

            for tool_call in tool_calls:
                function = tool_call["function"]
                tool_name = function["name"]

                try:
                    arguments = json.loads(
                        function["arguments"]
                    )
                except json.JSONDecodeError:
                    result = {
                        "success": False,
                        "retryable": False,
                        "error": "模型返回的工具参数不是合法 JSON"
                    }
                else:
                    result = execute_tool(
                        tool_name,
                        arguments
                    )

                print(
                    f"[Round {round_number + 1}] "
                    f"Tool: {tool_name}"
                )
                print(
                    "[Tool Result]",
                    json.dumps(result, ensure_ascii=False)
                )

                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call["id"],
                    "content": json.dumps(
                        result,
                        ensure_ascii=False
                    )
                })

        return "Agent 已达到最大执行轮数，任务未能完成。"

这里有几个值得注意的地方。

1. 工具调用仍然由 Python 执行。

大模型只负责决定调用哪个工具以及提供参数，不会直接访问测试环境。

2. HTTP 响应会进入 Agent 上下文。

例如，查询工具返回：

{
  "success": true,
  "retryable": false,
  "http_status": 200,
  "data": {
    "order_id": "ORDER1001",
    "amount": 199.0,
    "status": "paid",
    "user_id": "USER001"
  }
}

Agent 可以根据这份响应数据，继续选择金额和状态断言工具。

3. 断言工具不再依赖隐藏的全局数据。

订单数据通过工具参数传递，便于追踪每次断言使用的实际响应，也方便后续接入多个接口和不同测试数据。

4. max_rounds 仍然是必要的。

模型可能连续调用工具，也可能因为上下文不完整而反复请求。限制执行轮数可以避免 Agent 无限循环。

本示例为了突出 HTTP 接口接入，使用了简化的 Agent Loop。第 06 篇的任务状态持久化和断点恢复仍然属于独立模块；如果要将两者合并成完整项目，需要把本篇的 execute_tool() 接入原有 TaskManager 的执行入口，而不是绕过任务管理器。

六、编写项目入口

新建 main.py：

from agent import AITestAgent


def main():
    agent = AITestAgent(max_rounds=8)

    task = """
    请测试订单 ORDER1001：

    1. 查询订单详情。
    2. 检查订单金额是否为 199。
    3. 检查订单状态是否为 paid。
    4. 汇总测试结果。
    """

    result = agent.run(task)

    print("\n========== Agent Final Result ==========")
    print(result)


if __name__ == "__main__":
    main()

requirements.txt：

# 本篇 HTTP 服务和客户端使用 Python 标准库
# 无第三方依赖

目前项目只需要 Python 3.9 或更高版本，不需要额外安装第三方包。

七、运行测试，观察 Agent 的执行过程

现在启动两个终端。

终端一：启动 HTTP 测试服务

python mock_server.py

终端二：配置模型并启动 Agent

export LLM_API_KEY="你的API_KEY"
export LLM_BASE_URL="https://你的模型服务地址/v1"
export LLM_MODEL="你的模型名称"

python main.py

下面是一次正常测试的预期执行过程。具体工具调用轮数和日志格式会受模型服务影响。

[Round 1] Tool: query_order_detail

[Tool Result]
{
  "success": true,
  "retryable": false,
  "http_status": 200,
  "data": {
    "order_id": "ORDER1001",
    "amount": 199.0,
    "status": "paid",
    "user_id": "USER001"
  }
}

[Round 2] Tool: check_order_amount

[Tool Result]
{
  "success": true,
  "passed": true,
  "actual": 199.0,
  "expected": 199,
  "message": "订单金额符合预期"
}

[Round 3] Tool: check_order_status

[Tool Result]
{
  "success": true,
  "passed": true,
  "actual": "paid",
  "expected": "paid",
  "message": "订单状态符合预期"
}

最终 Agent 可能给出类似这样的总结：

订单 ORDER1001 测试完成。

接口调用：
- 订单查询接口：HTTP 200，请求成功。

业务断言：
- 订单金额：199，符合预期。
- 订单状态：paid，符合预期。

测试结论：通过。

以上为预期示例，不是已经实际调用模型服务得到的运行日志。

八、再测一个异常订单：HTTP 200 也可能测试失败

正常订单只能说明基本链路可以工作，还需要验证异常数据。

将 main.py 中的任务改成：

task = """
请测试订单 ORDER1002：

1. 查询订单详情。
2. 检查订单金额是否为 199。
3. 检查订单状态是否为 paid。
4. 汇总测试结果。
"""

预期查询结果：

{
  "success": true,
  "retryable": false,
  "http_status": 200,
  "data": {
    "order_id": "ORDER1002",
    "amount": -20.0,
    "status": "unknown",
    "user_id": "USER002"
  }
}

接下来两个业务断言应该都不通过：

{
  "success": true,
  "passed": false,
  "actual": -20.0,
  "expected": 199,
  "message": "订单金额不符合预期"
}

{
  "success": true,
  "passed": false,
  "actual": "unknown",
  "expected": "paid",
  "message": "订单状态不符合预期"
}

这里有一个非常重要的工程判断：

工具执行成功，不等于测试通过。

这两个断言工具都正常执行了，所以 success=true；但实际业务数据不符合预期，所以 passed=false。

如果把两者混在一起，后续测试报告就很容易出现错误统计。

九、接口异常应该如何处理？

接入 HTTP 接口以后，异常类型会比本地函数复杂得多。

本篇准备了三种异常场景，可以通过修改订单 ID 验证。

测试输入	HTTP 结果	是否重试	处理方式
NOT_FOUND	404	否	记录接口资源不存在
SERVER_ERROR	500	有限重试	达到上限后记录接口失败
SLOW	请求超时	有限重试	达到上限后记录超时

1. 404 不应该盲目重试

例如：

GET /api/orders/NOT_FOUND

如果订单确实不存在，重复请求通常不会改变结果。

因此应该把它记录为接口调用失败，并结束依赖该订单数据的后续断言。

当然，在某些最终一致性业务中，资源可能刚创建还未同步完成，此时 404 是否重试应由具体业务决定，而不是一概而论。

2. 500 可以有限重试

服务器内部错误可能是临时故障，也可能是持续性故障。

因此可以设置最大重试次数和退避间隔，但不能无限重试。

第 06 篇已经介绍了重试和异常恢复，本篇是在 HTTP 请求层落实这套思路。

3. 超时需要明确处理

超时不等于服务端一定没有执行请求。

尤其是 POST、支付、订单创建等具有副作用的接口，客户端超时后，服务端可能已经完成操作，只是响应没有及时返回。

这种情况下直接重试，可能产生重复数据。

本篇的订单查询使用 GET，适合做有限重试。涉及写操作时，还需要结合幂等键、请求状态查询等机制，不能简单复制查询接口的重试逻辑。

十、从 Demo 走向真实测试环境，还需要补什么？

到这里，我们已经完成了从本地模拟工具到 HTTP 接口工具的过渡。不过，示例项目和生产环境仍然有一些差距。

我认为至少需要关注以下几点：

1. 环境配置不能写死

测试环境、预发布环境和生产环境应使用不同的配置。尤其要避免让 Agent 在没有限制的情况下访问生产环境。

2. 接口鉴权信息不能暴露给模型

Token、Cookie、密钥等应由 Python 工具层读取环境变量或安全配置，不能放进模型提示词，也不要完整打印到执行日志。

3. 需要完善请求和响应校验

除了 HTTP 状态码，还应该校验：

• 响应是否为合法 JSON。
• 必要字段是否存在。
• 字段类型是否符合约定。
• 业务错误码是否符合预期。
• 响应结构是否发生变化。

4. 工具权限必须受控

不要因为 Agent 接入了 HTTP，就允许它任意访问接口。

应该明确工具白名单、请求方法、可访问路径和参数范围。涉及写操作的工具还应增加权限校验和风险控制。

5. 记录完整的执行链路

后续如果一个业务测试失败，我们需要知道：

• Agent 接收了什么任务。
• 模型选择了哪个工具。
• 实际发送了什么请求。
• 接口返回了什么结果。
• 执行了哪些断言。
• 哪一步发生了异常。

这些信息对于定位问题非常重要。

十一、本篇总结

本篇完成了 AI Test Agent 接入 HTTP 测试环境的基础能力。

模块	本篇实现
HTTP 测试服务	本地订单 API
接口调用	Python 标准库 HTTP 客户端
超时控制	2 秒请求超时
失败重试	有限次数和退避等待
工具注册	HTTP 查询工具加入 Agent
业务检查	金额、状态断言
结果区分	工具执行结果与业务断言结果分离

从这篇开始，Agent 已经不再只是调用本地模拟函数，而是可以通过受控的 HTTP 工具访问测试环境，并根据接口响应执行检查。

但还有一个问题没有解决：

现在测试任务仍然需要测试人员提前告诉 Agent，要检查哪个订单、检查哪些字段以及预期结果是什么。

能不能让 Agent 直接读取一段业务需求，自动分析测试点、生成测试用例，再调用刚刚接入的 HTTP 工具执行？

这就是下一篇要解决的问题。