> ## 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.

# Invoking Expert Agents

> Integrate Profy Expert into your application — invoke domain-specific AI agents via the Events API and handle SSE streaming responses.

This tutorial walks you through the complete process of invoking a Profy Expert from an external application — from obtaining the Expert identifier to building a full streaming chat interface.

## What You'll Build

A backend service that can invoke Profy Experts, with SSE streaming responses connected to a frontend chat interface. When complete, your application will have:

* The ability to invoke any published Expert
* Multi-turn conversation context retention
* SSE stream parsing and error handling
* Automatic per-token billing (METERED model)

## What Is an Expert

An Expert is an AI Agent product on the Profy platform, published by creators on the Marketplace. Each Expert consists of:

| Component | Description                                                                   |
| --------- | ----------------------------------------------------------------------------- |
| Persona   | Character and behavioral style definition                                     |
| Tools     | Callable tool set (search, code execution, file operations, etc.)             |
| Memory    | Cross-session memory that continuously accumulates user preferences and facts |
| Skills    | Reusable structured skills that define the Agent's behavioral patterns        |

Through the Events API's `/openapi/v1/events/invoke` endpoint, your application can invoke these Experts just like calling an API.

## Prerequisites

Before getting started, make sure you have:

* **Profy App**: Created and configured with OAuth in [Studio](https://app.profy.cn/studio)
* **OAuth Access Token**: Obtained through the authorization code flow (see [SDK Quick Start](/en/developers/quickstart))
* **Target Expert Identifier**: The unique identifier of the Expert you want to invoke

<Warning>
  `POST /openapi/v1/events/invoke` requires the `events:write` OAuth scope. Make sure your App requested this permission during creation.
</Warning>

## Step 1: Find the Expert Identifier

Every published Expert has a unique `identifier` (slug format) used for API calls.

How to find it:

1. **Marketplace page** — Open the Expert's detail page; the last path segment in the URL is the identifier, e.g., `https://app.profy.cn/expert/data-analyst` → `data-analyst`
2. **Studio** — If you're the Expert's creator, the identifier is shown in the basic information section of the editing page

<Tip>
  Store the Expert Identifier in environment variables or configuration files to avoid hardcoding.
</Tip>

## Step 2: Single-Turn Invocation

The core of invoking an Expert is sending a POST request to `/openapi/v1/events/invoke`, which returns an SSE stream.

<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", "Analyze the recent sales trends");
  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", "Analyze the recent sales trends")
  )
  print(answer)
  ```
</CodeGroup>

## Step 3: Multi-Turn Conversations

By passing a `session_id` parameter, the Expert continues the conversation within the same session context, retaining previous message history and memory.

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

  const sessionId = randomUUID();

  const answer1 = await invokeExpertWithSession(
    token.accessToken,
    "data-analyst",
    "Help me analyze the Q1 user growth data",
    sessionId
  );

  const answer2 = await invokeExpertWithSession(
    token.accessToken,
    "data-analyst",
    "What changed compared to Q4?",
    sessionId  // Same session_id — the Expert remembers the previous turn
  );

  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",
      "Help me analyze the Q1 user growth data",
      session_id,
  )

  answer2 = await invoke_expert_with_session(
      access_token, "data-analyst",
      "What changed compared to Q4?",
      session_id,  # Same session_id — the Expert remembers the previous turn
  )

  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>
  The `session_id` is generated and managed by your application. All calls under the same `session_id` share conversation context. A new `session_id` starts a fresh conversation.
</Note>

## Step 4: Handling the SSE Stream

An Expert's response is a standard SSE (Server-Sent Events) stream containing the following event types:

| Event             | Description                 | Data Fields                         |
| ----------------- | --------------------------- | ----------------------------------- |
| `text`            | Text content fragment       | `content: string`                   |
| `tool_call_chunk` | Tool invocation information | `name: string`, `arguments: string` |
| `complete`        | Conversation turn complete  | —                                   |
| `error`           | Runtime error               | `message: string`, `code: string`   |

Here's a generic SSE stream parser:

<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: "Generate a quarterly report",
    }),
  });

  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--- Complete ---");
        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": "Generate a quarterly report",
          },
      ) 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--- Complete ---")
                  case "error":
                      print(f"\n[Error] {event.code}: {event.message}")
  ```
</CodeGroup>

## Step 5: Error Handling

The invoke endpoint may return the following errors that your application should handle accordingly:

| HTTP Status | Meaning                          | Handling                                                |
| ----------- | -------------------------------- | ------------------------------------------------------- |
| 400         | Invalid request parameters       | Check `expert_identifier` and `message` fields          |
| 401         | Token invalid or expired         | Refresh the Access Token and retry                      |
| 403         | User lacks access to this Expert | Guide the user to purchase/subscribe on the Marketplace |
| 404         | Expert not found                 | Verify the identifier is correct                        |
| 402         | Insufficient user credits        | Prompt the user to top up                               |
| 502         | Agent Runtime error              | Retry later; contact the platform if it persists        |

<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>
  A 403 typically means the user hasn't purchased this Expert. For METERED-type Experts, the user needs to subscribe on the Marketplace first; for ONE\_TIME-type, the user needs to complete a one-time purchase.
</Warning>

## Step 6: Building a Chat Interface

A minimal example of connecting the SSE stream to a frontend chat interface:

<CodeGroup>
  ```typescript React Component 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="Type a message..."
          disabled={loading}
        />
      </div>
    );
  }
  ```

  ```python Python Terminal Chat 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"Starting conversation with {expert_identifier} (type 'quit' to exit)\n")

      while True:
          user_input = input("You: ")
          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[Error] {data.get('message', '')}")
                      elif data["type"] == "complete":
                          print()

          print()

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

<Note>
  The frontend doesn't call the Profy API directly — requests go through your backend proxy (`/api/expert/invoke`), which holds the OAuth Token and forwards the SSE stream. This avoids exposing the Access Token on the client side.
</Note>

## Complete Example

A runnable backend service that wraps Expert invocation as an API endpoint:

<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 Invocation vs AI Model Calls

The Profy Events API provides two types of AI calls. Choose the right endpoint for your scenario:

| Dimension             | Expert Invocation (`/events/invoke`)                               | AI Model Call (`/events/chat`)                      |
| --------------------- | ------------------------------------------------------------------ | --------------------------------------------------- |
| Target                | Specific Expert (with persona + tools + memory)                    | General AI model (OpenAI-compatible)                |
| Request Format        | `{ expert_identifier, message }`                                   | `{ model, messages }`                               |
| Context Management    | Automatically managed by the platform (`session_id`)               | Application maintains the `messages` array          |
| Tool Calls            | Expert comes with built-in tools, orchestrated by the platform     | Application defines and handles tools               |
| Memory                | Built-in cross-session memory                                      | No memory — each call is independent                |
| Use Cases             | Domain expert tasks (data analysis, customer service, writing)     | General text generation, translation, summarization |
| Billing               | METERED (per token)                                                | METERED (per token)                                 |
| Persona Customization | Pre-defined by the creator — no configuration needed by the caller | Application defines via system prompt               |

<Tip>
  If you need a "ready-to-use domain expert," use Expert invocation. If you need "raw model capabilities" with your own orchestration, use AI model calls. Both can be used together in the same application.
</Tip>

## Next Steps

<CardGroup cols={2}>
  <Card title="SDK Quick Start" icon="rocket" href="/en/developers/quickstart">
    Install the SDK, set up OAuth, and make your first event report
  </Card>

  <Card title="AI Model Calls" icon="brain" href="/en/developers/cookbook/ai-metered-billing">
    Use the OpenAI-compatible interface to call platform AI models
  </Card>

  <Card title="App Integration Tutorial" icon="puzzle-piece" href="/en/developers/quickstart">
    Create an App from scratch, configure OAuth, and publish to the Marketplace
  </Card>

  <Card title="API Reference" icon="code" href="/en/api/agents-run">
    Complete field documentation for the Events Invoke endpoint
  </Card>
</CardGroup>