# 最佳实践

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

## 场景 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/json` | JSON 请求体 | 视接口 |

### Step 1: 创建身份主体（Principal）

POST /api/principal/create — 为每个最终用户/场景创建独立身份主体

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| principalName | string | 主体名称（必填） |
| principalDescription | string | 主体描述 |

```bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/principal/create' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'Content-Type: application/json' \
  -d '{"principalName": "客服机器人主体"}'
```

### Step 2: 创建智能体（Agent）

POST /api/agent/create — 创建智能体实例，返回 agentId

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| agentName | string | 智能体名称（必填） |
| agentSoul | string | System Prompt |
| defaultModelName | string | 默认模型 |

```bash
curl -X POST 'https://router.fis.aliyuncs.com/finx/api/agent/create' \
  -H 'Authorization: Bearer <TOKEN>' \
  -H 'Content-Type: application/json' \
  -d '{"agentName": "我的智能体"}'
```

### Step 3: 提交对话（异步运行）

POST /api/agent/run/async — 立即返回 requestId，不阻塞

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| agentId | int64 | 智能体 ID（必填） |
| userInput | string | 发送给智能体的消息（必填） |
| sessionId | int64 | 会话 ID（不传则新建） |
| triggerSource | object | 触发来源，建议 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"}}'
```

### Step 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 事件结构](/docs/quickstart/data-model.md#sse-event-types)。

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

> 文件：`agent_conversation.py`

```python
# 公共：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"]
```

```python
# 步骤 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"]
```

```python
# 步骤 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"]
```

```python
# 步骤 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"]
```

```python
# 步骤 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)
```

## 场景 02 · 附件上传与引用

上传文件并在智能体对话中引用，获取处理结果。附件必须归属于会话。

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

### Step 1: 预创建空会话

POST /api/session/create

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| agentId | int64 | 智能体 ID（必填） |
| title | string | 会话标题（必填） |

```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财务分析"}'
```

### Step 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'
```

### Step 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>]}'
```

### Step 4: 消费结果

同「发起对话」的 SSE 消费流程，用 requestId 订阅事件流。

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

> 文件：`attachment_flow.py`

```python
# 步骤 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"]
```

```python
# 步骤 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"]
```

```python
# 步骤 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"]
```

```python
# 步骤 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)
```

## 场景 03 · 接入外部数据源

为智能体挂载外部服务，扩展调用能力。

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

## 路径 A：导入 MCP 服务

### Step 1: 导入 MCP 服务

POST /api/agent/finx/importMcpService — 支持标准 MCP JSON 格式

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| agentId | int64 | 智能体 ID（必填） |
| mcpServers | object | key=服务名, value=配置（仅 1 个条目） |
| autoGenerateDescription | bool | 自动生成描述 |

```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 服务

### Step 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"}}'
```

### Step 2: 为服务添加工具

POST /api/agent/finx/createTool — 每个工具对应一个 HTTP 端点

| 字段 | 说明 |
| --- | --- |
| toolName | 工具名称（必填） |
| description | 工具描述（模型据此决定何时调用） |
| parameterSchema | 入参 JSON Schema |
| httpEndpoint | HTTP 端点配置（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":"查询客户详情"}'
```

## 配置鉴权（按需）

### Step 1: 查询授权 Schema

POST /api/agent/finx/getAuthSchema — 获取需要填写的鉴权字段

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| agentId | int64 | 智能体 ID（必填） |
| serviceName | string | 服务名称（必填） |

```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"}'
```

### Step 2: 保存授权凭证

POST /api/agent/finx/saveAuthCredential — 提交凭证键值对

| 字段 | 类型 | 说明 |
| --- | --- | --- |
| agentId | int64 | 智能体 ID（必填） |
| serviceName | string | 服务名称（必填） |
| credentials | object | 凭证键值对（必填），例如 `{"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`

```python
# 配置流程：创建服务 → 鉴权 → 启用 → 校验（MCP 推荐 / HTTP 可选）
# 两种接入方式：MCP（推荐）或 HTTP，均按 agentId 维度配置。
```

```python
# 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},
        },
    ))
```

```python
# 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,
        },
    ))
```

```python
# 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,
        },
    ))
```

```python
# 鉴权 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,
        },
    ))
```

```python
# 鉴权 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,
        },
    ))
```

```python
# 启用与校验
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"))
```