指南 03 · HTTP 调用¶
让 Agent 去调第三方接口（业务系统、外部 API），并根据响应决定后续分支。对应节点 HttpRequest。
1. 基本用法¶
from agentframework import Workflow
from agentframework.nodes import Answer, HttpRequest
wf = Workflow(name="http_demo")
wf.add_node(
HttpRequest(
name="api",
url=wf.ref("input.url"), # 地址可来自工作流入口
method="GET", # GET/POST/PUT/DELETE/PATCH/HEAD
headers={"Authorization": "Bearer xxx"},
)
)
wf.add_node(Answer(name="ans", text=wf.ref("api.text"))) # 引用响应文本
wf.add_edge("api", "ans")
输入输出：
| 输入 | 类型 | 说明 |
|---|---|---|
url |
str 必填 |
请求地址 |
method |
str 可选 |
默认 GET |
headers |
dict 可选 |
请求头 |
body |
any 可选 |
请求体（dict/list 自动转 JSON；str 原样） |
| 输出 | 类型 | 说明 |
|---|---|---|
status_code |
int |
HTTP 状态码 |
text |
str |
响应文本 |
json |
dict |
响应 JSON（解析失败为 None） |
要点：
headers/body支持输入（如wf.ref）与节点参数互备。body传 dict/list 自动 JSON 化；网络错误包装为RuntimeError。- 可注入
client=httpx.Client（测试用httpx.MockTransport），不传则内部自建httpx.Client(timeout=30)。
2. POST + 按响应分流¶
调第三方接口拿 JSON，用 IfElse 判定成败后走不同分支：
from agentframework.nodes import IfElse, QuestionClassify
wf = Workflow(name="api_flow")
wf.add_node(
HttpRequest(
name="http",
url="http://localhost:9080/test1",
method="POST",
body={"flag": wf.ref("flag.branch")}, # 上游算好的值拼进请求体
)
)
# 把 IfElse 当判定节点：读 http.json 里的业务码
wf.add_node(
IfElse(
name="ok",
left=wf.ref("http.json"),
condition=lambda inputs, ctx: dict(inputs["left"]).get("code") == "FBASE0000",
)
)
wf.add_node(Answer(name="success", text="办理成功"))
wf.add_node(Answer(name="fail", text="办理失败，请稍后再试"))
wf.add_edge("http", "ok")
wf.add_conditional_edge("ok", {True: "success", False: "fail"})
3. 完整可运行示例¶
本站收录了完整可运行源码与运行说明：
- http_flow 示例：GET httpbin → Answer 回显响应文本。
- withdrawal_customer_service 示例：POST 第三方 + 按
code成败分流 + 失败读respMsg给用户的真实对接（含--mock-third-party可离线模拟第三方）。