> ## Documentation Index
> Fetch the complete documentation index at: https://docs.profy.cn/llms.txt
> Use this file to discover all available pages before exploring further.

# 调用 Expert 智能体

> 在你的应用中集成 Profy Expert，通过 Events API 调用领域智能体并处理 SSE 流式响应。

# 调用 Expert 智能体

本教程将带你完成从零开始在外部应用中调用 Profy Expert 的全流程——从获取 Expert 标识到构建完整的流式聊天界面。

## 你将构建什么

一个可以调用 Profy Expert 的后端服务，并将 SSE 流式响应接入前端聊天界面。完成后，你的应用将具备：

* 调用任意已发布 Expert 的能力
* 多轮对话上下文维持
* SSE 流式解析与错误处理
* 按 token 用量自动计费（METERED 模型）

## Expert 是什么

Expert 是 Profy 平台上的 AI Agent 产品，由创作者在 Marketplace 上发布。每个 Expert 包含：

| 组成      | 说明                       |
| ------- | ------------------------ |
| Persona | 人设与行为风格定义                |
| Tools   | 可调用的工具集（搜索、代码执行、文件操作等）   |
| Memory  | 跨会话记忆，持续积累用户偏好与事实        |
| Skills  | 可复用的结构化技能，定义 Agent 的行为模式 |

通过 Events API 的 `/openapi/v1/events/invoke` 端点，你的应用可以像调用一个 API 一样调用这些 Expert。

## 前置条件

在开始之前，确保你已具备：

* **Profy App**：在 [Studio](https://app.profy.cn/studio) 中创建并配置好 OAuth
* **OAuth Access Token**：通过授权码流程获取（参见 [SDK 快速开始](/zh/sdk/quickstart)）
* **目标 Expert Identifier**：你要调用的 Expert 的唯一标识

<Warning>
  `POST /openapi/v1/events/invoke` 需要 `events:write` OAuth scope。确保你的 App 在创建时申请了该权限。
</Warning>

## Step 1: 找到 Expert Identifier

每个已发布的 Expert 都有一个唯一的 `identifier`（slug 格式），用于 API 调用。

获取方式：

1. **Marketplace 页面** — 打开 Expert 详情页，URL 中的最后一段路径即为 identifier，如 `https://app.profy.cn/expert/data-analyst` → `data-analyst`
2. **Studio** — 如果你是 Expert 的创作者，在编辑页面的基本信息中可以看到 identifier

<Tip>
  建议将 Expert Identifier 存入环境变量或配置文件，避免硬编码。
</Tip>

## Step 2: 单轮调用

调用 Expert 的核心是向 `/openapi/v1/events/invoke` 发送 POST 请求，返回 SSE 流。

<CodeGroup>
  ```typescript TypeScript theme={null}
  const BASE_URL = "https://app.profy.cn";

  async function invokeExpert(
    accessToken: string,
    expertIdentifier: string,
    message: string
  ): Promise<string> {
    const res = await fetch(`${BASE_URL}/openapi/v1/events/invoke`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${accessToken}`,
      },
      body: JSON.stringify({
        expert_identifier: expertIdentifier,
        message,
      }),
    });

    if (!res.ok) {
      throw new Error(`Invoke failed: ${res.status} ${res.statusText}`);
    }

    let fullText = "";
    const reader = res.body!.getReader();
    const decoder = new TextDecoder();

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      const chunk = decoder.decode(value, { stream: true });
      for (const line of chunk.split("\n")) {
        if (line.startsWith("data: ")) {
          const data = JSON.parse(line.slice(6));
          if (data.type === "text") {
            fullText += data.content;
          }
        }
      }
    }

    return fullText;
  }

  const answer = await invokeExpert(token.accessToken, "data-analyst", "分析一下最近的销售趋势");
  console.log(answer);
  ```

  ```python Python theme={null}
  import httpx

  BASE_URL = "https://app.profy.cn"

  async def invoke_expert(
      access_token: str,
      expert_identifier: str,
      message: str,
  ) -> str:
      full_text = ""

      async with httpx.AsyncClient() as client:
          async with client.stream(
              "POST",
              f"{BASE_URL}/openapi/v1/events/invoke",
              headers={
                  "Content-Type": "application/json",
                  "Authorization": f"Bearer {access_token}",
              },
              json={
                  "expert_identifier": expert_identifier,
                  "message": message,
              },
          ) as response:
              response.raise_for_status()

              async for line in response.aiter_lines():
                  if line.startswith("data: "):
                      import json
                      data = json.loads(line[6:])
                      if data["type"] == "text":
                          full_text += data["content"]

      return full_text

  import asyncio
  answer = asyncio.run(
      invoke_expert(access_token, "data-analyst", "分析一下最近的销售趋势")
  )
  print(answer)
  ```
</CodeGroup>

## Step 3: 多轮对话

通过传递 `session_id` 参数，Expert 会在同一会话上下文中延续对话，保留之前的消息历史和记忆。

<CodeGroup>
  ```typescript TypeScript theme={null}
  import { randomUUID } from "crypto";

  const sessionId = randomUUID();

  const answer1 = await invokeExpertWithSession(
    token.accessToken,
    "data-analyst",
    "帮我分析一下 Q1 的用户增长数据",
    sessionId
  );

  const answer2 = await invokeExpertWithSession(
    token.accessToken,
    "data-analyst",
    "和 Q4 相比有什么变化？",
    sessionId  // 同一个 session_id，Expert 记得上一轮的内容
  );

  async function invokeExpertWithSession(
    accessToken: string,
    expertIdentifier: string,
    message: string,
    sessionId: string
  ): Promise<string> {
    const res = await fetch(`${BASE_URL}/openapi/v1/events/invoke`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${accessToken}`,
      },
      body: JSON.stringify({
        expert_identifier: expertIdentifier,
        message,
        session_id: sessionId,
      }),
    });

    if (!res.ok) {
      throw new Error(`Invoke failed: ${res.status}`);
    }

    let fullText = "";
    const reader = res.body!.getReader();
    const decoder = new TextDecoder();

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      const chunk = decoder.decode(value, { stream: true });
      for (const line of chunk.split("\n")) {
        if (line.startsWith("data: ")) {
          const data = JSON.parse(line.slice(6));
          if (data.type === "text") {
            fullText += data.content;
          }
        }
      }
    }

    return fullText;
  }
  ```

  ```python Python theme={null}
  import uuid

  session_id = str(uuid.uuid4())

  answer1 = await invoke_expert_with_session(
      access_token, "data-analyst",
      "帮我分析一下 Q1 的用户增长数据",
      session_id,
  )

  answer2 = await invoke_expert_with_session(
      access_token, "data-analyst",
      "和 Q4 相比有什么变化？",
      session_id,  # 同一个 session_id，Expert 记得上一轮的内容
  )

  async def invoke_expert_with_session(
      access_token: str,
      expert_identifier: str,
      message: str,
      session_id: str,
  ) -> str:
      full_text = ""

      async with httpx.AsyncClient() as client:
          async with client.stream(
              "POST",
              f"{BASE_URL}/openapi/v1/events/invoke",
              headers={
                  "Content-Type": "application/json",
                  "Authorization": f"Bearer {access_token}",
              },
              json={
                  "expert_identifier": expert_identifier,
                  "message": message,
                  "session_id": session_id,
              },
          ) as response:
              response.raise_for_status()

              async for line in response.aiter_lines():
                  if line.startswith("data: "):
                      import json
                      data = json.loads(line[6:])
                      if data["type"] == "text":
                          full_text += data["content"]

      return full_text
  ```
</CodeGroup>

<Note>
  `session_id` 由你的应用生成和管理。同一个 `session_id` 下的所有调用共享对话上下文。新的 `session_id` 会开启全新对话。
</Note>

## Step 4: 处理 SSE 流

Expert 的响应是标准 SSE（Server-Sent Events）流，包含以下事件类型：

| 事件                | 说明     | 数据字段                                |
| ----------------- | ------ | ----------------------------------- |
| `text`            | 文本内容片段 | `content: string`                   |
| `tool_call_chunk` | 工具调用信息 | `name: string`, `arguments: string` |
| `complete`        | 对话轮次结束 | —                                   |
| `error`           | 运行时错误  | `message: string`, `code: string`   |

下面是一个通用的 SSE 流解析器：

<CodeGroup>
  ```typescript TypeScript theme={null}
  interface ExpertEvent {
    type: "text" | "tool_call_chunk" | "complete" | "error";
    content?: string;
    name?: string;
    arguments?: string;
    message?: string;
    code?: string;
  }

  async function* streamExpertEvents(
    response: Response
  ): AsyncGenerator<ExpertEvent> {
    const reader = response.body!.getReader();
    const decoder = new TextDecoder();
    let buffer = "";

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split("\n");
      buffer = lines.pop() ?? "";

      for (const line of lines) {
        if (line.startsWith("data: ")) {
          yield JSON.parse(line.slice(6)) as ExpertEvent;
        }
      }
    }
  }

  const response = await fetch(`${BASE_URL}/openapi/v1/events/invoke`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: `Bearer ${accessToken}`,
    },
    body: JSON.stringify({
      expert_identifier: "data-analyst",
      message: "生成季度报告",
    }),
  });

  for await (const event of streamExpertEvents(response)) {
    switch (event.type) {
      case "text":
        process.stdout.write(event.content ?? "");
        break;
      case "tool_call_chunk":
        console.log(`\n[Tool] ${event.name}(${event.arguments})`);
        break;
      case "complete":
        console.log("\n--- 完成 ---");
        break;
      case "error":
        console.error(`\n[Error] ${event.code}: ${event.message}`);
        break;
    }
  }
  ```

  ```python Python theme={null}
  from dataclasses import dataclass
  from typing import AsyncIterator, Optional
  import json

  @dataclass
  class ExpertEvent:
      type: str
      content: Optional[str] = None
      name: Optional[str] = None
      arguments: Optional[str] = None
      message: Optional[str] = None
      code: Optional[str] = None

  async def stream_expert_events(
      response: httpx.Response,
  ) -> AsyncIterator[ExpertEvent]:
      async for line in response.aiter_lines():
          if line.startswith("data: "):
              data = json.loads(line[6:])
              yield ExpertEvent(**data)

  async with httpx.AsyncClient() as client:
      async with client.stream(
          "POST",
          f"{BASE_URL}/openapi/v1/events/invoke",
          headers={
              "Content-Type": "application/json",
              "Authorization": f"Bearer {access_token}",
          },
          json={
              "expert_identifier": "data-analyst",
              "message": "生成季度报告",
          },
      ) as response:
          response.raise_for_status()

          async for event in stream_expert_events(response):
              match event.type:
                  case "text":
                      print(event.content, end="", flush=True)
                  case "tool_call_chunk":
                      print(f"\n[Tool] {event.name}({event.arguments})")
                  case "complete":
                      print("\n--- 完成 ---")
                  case "error":
                      print(f"\n[Error] {event.code}: {event.message}")
  ```
</CodeGroup>

## Step 5: 错误处理

invoke 端点可能返回以下错误，你的应用需要针对性处理：

| HTTP 状态码 | 含义               | 处理方式                                  |
| -------- | ---------------- | ------------------------------------- |
| 400      | 请求参数错误           | 检查 `expert_identifier` 和 `message` 字段 |
| 401      | Token 无效或过期      | 刷新 Access Token 后重试                   |
| 403      | 用户无权访问该 Expert   | 引导用户到 Marketplace 购买/订阅               |
| 404      | Expert 不存在       | 检查 identifier 是否正确                    |
| 402      | 用户积分不足           | 提示用户充值                                |
| 502      | Agent Runtime 错误 | 稍后重试，若持续则联系平台                         |

<CodeGroup>
  ```typescript TypeScript theme={null}
  async function safeInvoke(
    accessToken: string,
    expertIdentifier: string,
    message: string,
    sessionId?: string
  ): Promise<{ ok: true; text: string } | { ok: false; error: string }> {
    const res = await fetch(`${BASE_URL}/openapi/v1/events/invoke`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${accessToken}`,
      },
      body: JSON.stringify({
        expert_identifier: expertIdentifier,
        message,
        ...(sessionId && { session_id: sessionId }),
      }),
    });

    if (!res.ok) {
      switch (res.status) {
        case 401:
          return { ok: false, error: "TOKEN_EXPIRED" };
        case 402:
          return { ok: false, error: "INSUFFICIENT_BALANCE" };
        case 403:
          return { ok: false, error: "ACCESS_DENIED" };
        case 404:
          return { ok: false, error: "EXPERT_NOT_FOUND" };
        case 502:
          return { ok: false, error: "RUNTIME_ERROR" };
        default:
          return { ok: false, error: `UNKNOWN_${res.status}` };
      }
    }

    let text = "";
    for await (const event of streamExpertEvents(res)) {
      if (event.type === "text") text += event.content ?? "";
      if (event.type === "error") return { ok: false, error: event.message ?? "STREAM_ERROR" };
    }

    return { ok: true, text };
  }
  ```

  ```python Python theme={null}
  async def safe_invoke(
      access_token: str,
      expert_identifier: str,
      message: str,
      session_id: str | None = None,
  ) -> dict:
      payload = {
          "expert_identifier": expert_identifier,
          "message": message,
      }
      if session_id:
          payload["session_id"] = session_id

      async with httpx.AsyncClient() as client:
          try:
              async with client.stream(
                  "POST",
                  f"{BASE_URL}/openapi/v1/events/invoke",
                  headers={
                      "Content-Type": "application/json",
                      "Authorization": f"Bearer {access_token}",
                  },
                  json=payload,
              ) as response:
                  if response.status_code == 401:
                      return {"ok": False, "error": "TOKEN_EXPIRED"}
                  if response.status_code == 402:
                      return {"ok": False, "error": "INSUFFICIENT_BALANCE"}
                  if response.status_code == 403:
                      return {"ok": False, "error": "ACCESS_DENIED"}
                  if response.status_code == 404:
                      return {"ok": False, "error": "EXPERT_NOT_FOUND"}
                  if response.status_code == 502:
                      return {"ok": False, "error": "RUNTIME_ERROR"}

                  response.raise_for_status()

                  text = ""
                  async for event in stream_expert_events(response):
                      if event.type == "text":
                          text += event.content or ""
                      if event.type == "error":
                          return {"ok": False, "error": event.message or "STREAM_ERROR"}

                  return {"ok": True, "text": text}

          except httpx.HTTPStatusError:
              return {"ok": False, "error": f"UNKNOWN_{response.status_code}"}
  ```
</CodeGroup>

<Warning>
  403 通常意味着用户尚未购买该 Expert。对于 METERED 类型的 Expert，用户需要先在 Marketplace 完成订阅；对于 ONE\_TIME 类型，用户需要完成一次性购买。
</Warning>

## Step 6: 构建聊天界面

将 SSE 流接入前端聊天界面的最小示例：

<CodeGroup>
  ```typescript React 组件 theme={null}
  "use client";

  import { useState, useCallback } from "react";

  interface Message {
    role: "user" | "assistant";
    content: string;
  }

  export function ExpertChat({ expertId }: { expertId: string }) {
    const [messages, setMessages] = useState<Message[]>([]);
    const [input, setInput] = useState("");
    const [loading, setLoading] = useState(false);
    const [sessionId] = useState(() => crypto.randomUUID());

    const send = useCallback(async () => {
      if (!input.trim() || loading) return;

      const userMsg: Message = { role: "user", content: input };
      setMessages((prev) => [...prev, userMsg]);
      setInput("");
      setLoading(true);

      const assistantMsg: Message = { role: "assistant", content: "" };
      setMessages((prev) => [...prev, assistantMsg]);

      try {
        const res = await fetch("/api/expert/invoke", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            expertId,
            message: input,
            sessionId,
          }),
        });

        const reader = res.body!.getReader();
        const decoder = new TextDecoder();

        while (true) {
          const { done, value } = await reader.read();
          if (done) break;

          const chunk = decoder.decode(value, { stream: true });
          for (const line of chunk.split("\n")) {
            if (!line.startsWith("data: ")) continue;
            const data = JSON.parse(line.slice(6));
            if (data.type === "text") {
              setMessages((prev) => {
                const updated = [...prev];
                const last = updated[updated.length - 1];
                updated[updated.length - 1] = {
                  ...last,
                  content: last.content + (data.content ?? ""),
                };
                return updated;
              });
            }
          }
        }
      } finally {
        setLoading(false);
      }
    }, [input, loading, expertId, sessionId]);

    return (
      <div>
        <div>
          {messages.map((msg, i) => (
            <div key={i} data-role={msg.role}>
              {msg.content}
            </div>
          ))}
        </div>
        <input
          value={input}
          onChange={(e) => setInput(e.target.value)}
          onKeyDown={(e) => e.key === "Enter" && send()}
          placeholder="输入消息..."
          disabled={loading}
        />
      </div>
    );
  }
  ```

  ```python Python 终端聊天 theme={null}
  import asyncio
  import uuid
  import httpx
  import json

  BASE_URL = "https://app.profy.cn"

  async def chat_loop(access_token: str, expert_identifier: str):
      session_id = str(uuid.uuid4())
      print(f"开始与 {expert_identifier} 对话（输入 quit 退出）\n")

      while True:
          user_input = input("你: ")
          if user_input.strip().lower() == "quit":
              break

          print("Expert: ", end="", flush=True)

          async with httpx.AsyncClient() as client:
              async with client.stream(
                  "POST",
                  f"{BASE_URL}/openapi/v1/events/invoke",
                  headers={
                      "Content-Type": "application/json",
                      "Authorization": f"Bearer {access_token}",
                  },
                  json={
                      "expert_identifier": expert_identifier,
                      "message": user_input,
                      "session_id": session_id,
                  },
                  timeout=60.0,
              ) as response:
                  response.raise_for_status()

                  async for line in response.aiter_lines():
                      if not line.startswith("data: "):
                          continue
                      data = json.loads(line[6:])
                      if data["type"] == "text":
                          print(data.get("content", ""), end="", flush=True)
                      elif data["type"] == "error":
                          print(f"\n[错误] {data.get('message', '')}")
                      elif data["type"] == "complete":
                          print()

          print()

  asyncio.run(chat_loop("your_access_token", "data-analyst"))
  ```
</CodeGroup>

<Note>
  前端不直接调用 Profy API——请求经过你的后端代理（`/api/expert/invoke`），后端持有 OAuth Token 并转发 SSE 流。这避免了在前端暴露 Access Token。
</Note>

## 完整示例

一个可运行的后端服务，将 Expert 调用封装为 API 端点：

<CodeGroup>
  ```typescript server.ts theme={null}
  import { Hono } from "hono";
  import { stream } from "hono/streaming";

  const app = new Hono();
  const PROFY_BASE = "https://app.profy.cn";

  app.post("/api/expert/invoke", async (c) => {
    const { expertId, message, sessionId } = await c.req.json();
    const accessToken = await getAccessTokenForUser(c);

    const upstream = await fetch(`${PROFY_BASE}/openapi/v1/events/invoke`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${accessToken}`,
      },
      body: JSON.stringify({
        expert_identifier: expertId,
        message,
        session_id: sessionId,
      }),
    });

    if (!upstream.ok) {
      return c.json({ error: `upstream_${upstream.status}` }, upstream.status as 400);
    }

    return stream(c, async (s) => {
      const reader = upstream.body!.getReader();
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        await s.write(value);
      }
    });
  });

  export default app;
  ```

  ```python server.py theme={null}
  from fastapi import FastAPI, Request
  from fastapi.responses import StreamingResponse
  import httpx

  app = FastAPI()
  PROFY_BASE = "https://app.profy.cn"

  @app.post("/api/expert/invoke")
  async def invoke_expert(request: Request):
      body = await request.json()
      access_token = await get_access_token_for_user(request)

      async def proxy_stream():
          async with httpx.AsyncClient() as client:
              async with client.stream(
                  "POST",
                  f"{PROFY_BASE}/openapi/v1/events/invoke",
                  headers={
                      "Content-Type": "application/json",
                      "Authorization": f"Bearer {access_token}",
                  },
                  json={
                      "expert_identifier": body["expertId"],
                      "message": body["message"],
                      "session_id": body.get("sessionId"),
                  },
              ) as response:
                  response.raise_for_status()
                  async for chunk in response.aiter_bytes():
                      yield chunk

      return StreamingResponse(
          proxy_stream(),
          media_type="text/event-stream",
      )
  ```
</CodeGroup>

## Expert 调用 vs AI 模型调用

Profy Events API 提供两种 AI 调用方式。根据你的场景选择合适的端点：

| 维度    | Expert 调用 (`/events/invoke`)          | AI 模型调用 (`/events/chat`) |
| ----- | ------------------------------------- | ------------------------ |
| 调用对象  | 特定 Expert（含 persona + tools + memory） | 通用 AI 模型（OpenAI 兼容）      |
| 请求格式  | `{ expert_identifier, message }`      | `{ model, messages }`    |
| 上下文管理 | 平台自动管理（`session_id`）                  | 应用自行维护 `messages` 数组     |
| 工具调用  | Expert 自带工具，平台自动编排                    | 需要应用自行定义和处理              |
| 记忆    | Expert 内置跨会话记忆                        | 无记忆，每次调用独立               |
| 适用场景  | 领域专家任务（数据分析、客服、写作）                    | 通用文本生成、翻译、摘要             |
| 计费    | METERED（按 token）                      | METERED（按 token）         |
| 人设定制  | 创作者预定义，调用方无需配置                        | 应用自行通过 system prompt 定义  |

<Tip>
  如果你需要的是一个「开箱即用的领域专家」，用 Expert 调用；如果你需要的是「底层模型能力」并自行编排，用 AI 模型调用。两者可以在同一个应用中混合使用。
</Tip>

## 下一步

<CardGroup cols={2}>
  <Card title="SDK 快速开始" icon="rocket" href="/zh/sdk/quickstart">
    安装 SDK、完成 OAuth 对接和首次事件上报
  </Card>

  <Card title="AI 模型调用" icon="brain" href="/zh/sdk/cookbook/ai-metered-billing">
    使用 OpenAI 兼容接口调用平台 AI 模型
  </Card>

  <Card title="应用接入教程" icon="puzzle-piece" href="/zh/documentation/integration-quickstart">
    从零创建 App、配置 OAuth、上架市场
  </Card>

  <Card title="API 参考" icon="code" href="/zh/api/post-invoke">
    Events Invoke 端点完整字段说明
  </Card>
</CardGroup>