Finx
点金文档中心
查看 Markdown

最佳实践

典型场景的完整调用链与代码示例，覆盖从身份认证到获取结果的完整链路。每个场景均提供 curl 完整代码，可直接复制运行。

查看 Markdown
场景 01
发起智能体对话
从创建身份主体到实时消费智能体输出的端到端流程。

调用流程： 创建身份主体（Principal） → 创建智能体（Agent） → 异步提交对话（获取 requestId） → 订阅 SSE 事件流。

前置条件

已开通点金租户并创建应用、已获取 API Key。

text
Base URL: https://router.fis.aliyuncs.com/finx

鉴权说明（二要素）

Header说明必须
Authorization: Bearer <TOKEN>访问令牌（API Key）是
X-Principal-Id: <id>身份主体 ID运行类接口必须
Content-Type: application/jsonJSON 请求体视接口
1
创建身份主体（Principal）
POST /api/principal/create — 为每个最终用户/场景创建独立身份主体
字段类型说明
principalNamestring主体名称（必填）
principalDescriptionstring主体描述
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/principal/create' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'Content-Type: application/json' \
  -d '{"principalName": "客服机器人主体"}'
2
创建智能体（Agent）
POST /api/agent/create — 创建智能体实例，返回 agentId
字段类型说明
agentNamestring智能体名称（必填）
agentSoulstringSystem Prompt
defaultModelNamestring默认模型
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/create' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'Content-Type: application/json' \
  -d '{"agentName": "我的智能体"}'
3
提交对话（异步运行）
POST /api/agent/run/async — 立即返回 requestId，不阻塞
字段类型说明
agentIdint64智能体 ID（必填）
userInputstring发送给智能体的消息（必填）
sessionIdint64会话 ID（不传则新建）
triggerSourceobject触发来源，建议 api_call
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/run/async' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -H 'Content-Type: application/json' \
  -d '{"agentId":1001,"userInput":"你好","triggerSource":{"sourceType":"api_call"}}'
4
消费 SSE 事件流
GET /api/agent/run/stream?requestId=xxx — 实时读取智能体输出
参数说明
requestId步骤 3 返回的异步请求 ID（必填）
lastEventId断线重连时传入上次最大事件 ID
bash
curl -N -X GET 'https://router.fis.aliyuncs.com/finx/api/agent/run/stream?requestId=<requestId>' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'Accept: text/event-stream'

SSE 数据格式：每条事件的 event 为 EventType 枚举名（如 TEXT_DELTA），data 为完整的 AgentEvent JSON 信封（包含 eventType、data、timestamp、step、totalTokens）。

text
id: 0
event: TEXT_DELTA
data: {"eventType":"TEXT_DELTA","data":{"type":"text","text":"你好...","synthetic":false},"timestamp":"2026-05-07T10:30:01","step":1,"totalTokens":256}

更多事件类型和数据结构详见数据模型 -> SSE 事件结构。

API 接入必须携带 X-Principal-Id；提交对话后应尽快订阅 SSE；断线用 lastEventId 重连；所有非流式接口先判断 success 再取 data。

agent_conversation.py
# 公共：Base URL + Token + 鉴权头 import requests

BASE_URL = "https://router.fis.aliyuncs.com/finx"
TOKEN = "<YOUR_API_KEY>"

def auth_headers(principal_id=None, json_body=True):
h = {"Authorization": f"Bearer {TOKEN}"}
if json_body:
h["Content-Type"] = "application/json"
if principal_id is not None:
h["X-Principal-Id"] = str(principal_id)
return h

def unwrap(resp):
body = resp.json()
if not body.get("success"):
raise RuntimeError(body["message"])
return body["data"]
# 步骤 1：创建身份主体 → principalId def create_principal(name="客服机器人主体"):
data = unwrap(requests.post(
f"{BASE_URL}/api/principal/create",
headers=auth_headers(),
json={"principalName": name},
))
return data["principalId"]
# 步骤 2：创建智能体 → agentId def create_agent(principal_id, name="我的智能体"):
data = unwrap(requests.post(
f"{BASE_URL}/api/agent/create",
headers=auth_headers(principal_id),
json={"agentName": name},
))
return data["agentId"]
# 步骤 3：异步提交对话 → requestId def run_async(principal_id, agent_id, user_input):
data = unwrap(requests.post(
f"{BASE_URL}/api/agent/run/async",
headers=auth_headers(principal_id),
json={
"agentId": agent_id,
"userInput": user_input,
"triggerSource": {"sourceType": "api_call"},
},
))
return data["requestId"], data["sessionId"]
# 步骤 4：SSE 消费事件流 def stream_events(principal_id, request_id):
headers = auth_headers(principal_id, json_body=False)
headers["Accept"] = "text/event-stream"
with requests.get(
f"{BASE_URL}/api/agent/run/stream",
headers=headers,
params={"requestId": request_id},
stream=True, timeout=300,
) as resp:
for raw in resp.iter_lines(decode_unicode=True):
if raw and raw.startswith("data:"):
print(raw[5:].strip())

if __name__ == "__main__":
pid = create_principal()
aid = create_agent(pid)
req_id, sid = run_async(pid, aid, "你好，请做个自我介绍")
stream_events(pid, req_id)
查看 Markdown
场景 02
附件上传与引用
上传文件并在智能体对话中引用，获取处理结果。附件必须归属于会话。

配置流程： 先建会话 → 上传附件 → 带 attachmentIds 提交对话。

1
预创建空会话
POST /api/session/create
字段类型说明
agentIdint64智能体 ID（必填）
titlestring会话标题（必填）
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/session/create' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -H 'Content-Type: application/json' \
  -d '{"agentId":1001,"title":"Q3财务分析"}'
2
上传附件
POST /api/agent/attachment/upload（multipart/form-data）
响应字段说明
attachmentId附件 ID（提交对话时使用）
fileName原始文件名
fileSize文件大小（字节）
previewUrl预览 URL
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/attachment/upload?sessionId=<sid>' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -F 'file=@/path/to/report.pdf'
3
带附件提交对话
POST /api/agent/run/async — 与普通对话相比多了 sessionId + attachmentIds
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/run/async' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -H 'Content-Type: application/json' \
  -d '{"agentId":1001,"userInput":"帮我分析这份财报","sessionId":<sid>,"attachmentIds":[<attId>]}'
4
消费结果
同「发起对话」的 SSE 消费流程，用 requestId 订阅事件流。

attachmentIds 必须属于本会话且为当前用户上传；Content-Type 使用 multipart/form-data；附件上传前必须先有 sessionId。

attachment_flow.py
# 步骤 1：预创建会话 → sessionId def create_session(principal_id, agent_id, title="附件分析"):
data = unwrap(requests.post(
f"{BASE_URL}/api/session/create",
headers=auth_headers(principal_id),
json={"agentId": agent_id, "title": title},
))
return data["sessionId"]
# 步骤 2：上传附件（multipart）→ attachmentId def upload_attachment(principal_id, session_id, file_path):
with open(file_path, "rb") as f:
resp = requests.post(
f"{BASE_URL}/api/agent/attachment/upload",
headers=auth_headers(principal_id, json_body=False),
params={"sessionId": session_id},
files={"file": f},
)
return unwrap(resp)["attachmentId"]
# 步骤 3：带附件提交对话 def run_async_with_attachment(
principal_id, agent_id, sid, att_id, user_input,
):
data = unwrap(requests.post(
f"{BASE_URL}/api/agent/run/async",
headers=auth_headers(principal_id),
json={
"agentId": agent_id,
"userInput": user_input,
"sessionId": sid,
"attachmentIds": [att_id],
"triggerSource": {"sourceType": "api_call"},
},
))
return data["requestId"]
# 步骤 4：复用 stream_events 消费结果 if __name__ == "__main__":
pid = create_principal()
aid = create_agent(pid)
sid = create_session(pid, aid, "财报分析")
att = upload_attachment(pid, sid, "/path/to/report.pdf")
req_id = run_async_with_attachment(
pid, aid, sid, att, "帮我分析这份财报",
)
stream_events(pid, req_id)
查看 Markdown
场景 03
接入外部数据源
为智能体挂载外部服务，扩展调用能力。

配置流程： 创建服务 → （如需鉴权：查 Schema → 存凭证） → 启用服务 → 校验工具列表 → 对话中自动调用

路径 A：导入 MCP 服务

1
导入 MCP 服务
POST /api/agent/finx/importMcpService — 支持标准 MCP JSON 格式
字段类型说明
agentIdint64智能体 ID（必填）
mcpServersobjectkey=服务名, value=配置（仅 1 个条目）
autoGenerateDescriptionbool自动生成描述
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/finx/importMcpService' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -H 'Content-Type: application/json' \
  -d '{
    "agentId": 1001,
    "autoGenerateDescription": true,
    "mcpServers": {
      "my-weather": {
        "type": "sse",
        "url": "https://example.com/mcp/sse"
      }
    }
  }'

路径 B：创建 HTTP 服务

1
创建 HTTP 服务
POST /api/agent/finx/createHttpService
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/finx/createHttpService' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -H 'Content-Type: application/json' \
  -d '{"agentId":1001,"serviceName":"crm-api","httpConfig":{"baseUrl":"https://crm.example.com/api"}}'
2
为服务添加工具
POST /api/agent/finx/createTool — 每个工具对应一个 HTTP 端点
字段说明
toolName工具名称（必填）
description工具描述（模型据此决定何时调用）
parameterSchema入参 JSON Schema
httpEndpointHTTP 端点配置（method、path）
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/finx/createTool' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -H 'Content-Type: application/json' \
  -d '{"agentId":1001,"serviceId":5001,"toolName":"queryCustomer","description":"查询客户详情"}'

配置鉴权（按需）

1
查询授权 Schema
POST /api/agent/finx/getAuthSchema — 获取需要填写的鉴权字段
字段类型说明
agentIdint64智能体 ID（必填）
serviceNamestring服务名称（必填）
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/finx/getAuthSchema' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -H 'Content-Type: application/json' \
  -d '{"agentId":1001,"serviceName":"my-weather"}'
2
保存授权凭证
POST /api/agent/finx/saveAuthCredential — 提交凭证键值对
字段类型说明
agentIdint64智能体 ID（必填）
serviceNamestring服务名称（必填）
credentialsobject凭证键值对（必填），例如 {"apiKey": "xxx"}
bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/finx/saveAuthCredential' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'X-Principal-Id: <principalId>' \
  -H 'Content-Type: application/json' \
  -d '{"agentId":1001,"serviceName":"my-weather","credentials":{"apiKey":"sk-xxxx"}}'

启用与校验

操作方法路径
启用/禁用服务POST/api/agent/finx/toggleServiceEnabled
查询服务列表GET/api/agent/finx/listServices
查询工具列表GET/api/agent/finx/listTools
调试工具POST/api/agent/finx/execute

使用 POST /api/agent/finx/execute 直接调试工具，命令格式：finx <服务名> <工具名> [--参数名 值]

finx_tools.py
# 配置流程：创建服务 → 鉴权 → 启用 → 校验（MCP 推荐 / HTTP 可选） # 两种接入方式：MCP（推荐）或 HTTP，均按 agentId 维度配置。
# A. 导入 MCP 服务（最简，推荐） def import_mcp_service(principal_id, agent_id, name, mcp_cfg):
return unwrap(requests.post(
f"{BASE_URL}/api/agent/finx/importMcpService",
headers=auth_headers(principal_id),
json={
"agentId": agent_id,
"autoGenerateDescription": True,
"mcpServers": {name: mcp_cfg},
},
))
# B1. 创建 HTTP 服务 → serviceId def create_http_service(principal_id, agent_id, name, http_cfg):
return unwrap(requests.post(
f"{BASE_URL}/api/agent/finx/createHttpService",
headers=auth_headers(principal_id),
json={
"agentId": agent_id,
"serviceName": name,
"httpConfig": http_cfg,
},
))
# B2. 为 HTTP 服务添加工具 → toolId def create_tool(
principal_id, agent_id, sid, name, desc, schema, ep,
):
return unwrap(requests.post(
f"{BASE_URL}/api/agent/finx/createTool",
headers=auth_headers(principal_id),
json={
"agentId": agent_id,
"serviceId": sid,
"toolName": name,
"description": desc,
"parameterSchema": schema,
"httpEndpoint": ep,
},
))
# 鉴权 1：查询授权 Schema def get_auth_schema(principal_id, agent_id, service_name):
return unwrap(requests.post(
f"{BASE_URL}/api/agent/finx/getAuthSchema",
headers=auth_headers(principal_id),
json={
"agentId": agent_id,
"serviceName": service_name,
},
))
# 鉴权 2：保存授权凭证 def save_auth_credential(
principal_id, agent_id, service_name, creds,
):
return unwrap(requests.post(
f"{BASE_URL}/api/agent/finx/saveAuthCredential",
headers=auth_headers(principal_id),
json={
"agentId": agent_id,
"serviceName": service_name,
"credentials": creds,
},
))
# 启用与校验 def toggle_service(principal_id, agent_id, name, enabled=True):
return unwrap(requests.post(
f"{BASE_URL}/api/agent/finx/toggleServiceEnabled",
headers=auth_headers(principal_id),
json={
"agentId": agent_id,
"serviceName": name,
"enabled": enabled,
},
))

def list_tools(principal_id, agent_id, name):
return unwrap(requests.get(
f"{BASE_URL}/api/agent/finx/listTools",
headers=auth_headers(principal_id, json_body=False),
params={"agentId": agent_id, "serviceName": name},
))

if __name__ == "__main__":
pid = create_principal()
aid = create_agent(pid)
import_mcp_service(pid, aid, "my-weather", {
"type": "sse",
"url": "https://example.com/mcp/sse",
})
toggle_service(pid, aid, "my-weather", True)
print("已加载工具:", list_tools(pid, aid, "my-weather"))