Invoking Expert Agents

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
OAuth Access Token: Obtained through the authorization code flow (see SDK Quick Start)
Target Expert Identifier: The unique identifier of the Expert you want to invoke

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

Step 1: Find the Expert Identifier

Every published Expert has a unique identifier (slug format) used for API calls. How to find it:

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
Studio — If you’re the Expert’s creator, the identifier is shown in the basic information section of the editing page

Store the Expert Identifier in environment variables or configuration files to avoid hardcoding.

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.

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);

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)

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.

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;
}

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

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.

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:

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;
  }
}

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}")

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

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 };
}

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

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.

Step 6: Building a Chat Interface

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

"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>
  );
}

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"))

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.

Complete Example

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

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;

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",
    )

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

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.

Next Steps

SDK Quick Start

Install the SDK, set up OAuth, and make your first event report

AI Model Calls

Use the OpenAI-compatible interface to call platform AI models

App Integration Tutorial

Create an App from scratch, configure OAuth, and publish to the Marketplace

API Reference

Complete field documentation for the Events Invoke endpoint

Get Started

Authentication & Keys

API Guides

SDK

Integrations

Cookbook

Reference

Invoking Expert Agents

What You’ll Build

What Is an Expert

Prerequisites

Step 1: Find the Expert Identifier

Step 2: Single-Turn Invocation

Step 3: Multi-Turn Conversations

Step 4: Handling the SSE Stream

Step 5: Error Handling

Step 6: Building a Chat Interface

Complete Example

Expert Invocation vs AI Model Calls

Next Steps

SDK Quick Start

AI Model Calls

App Integration Tutorial

API Reference

​What You’ll Build

​What Is an Expert

​Prerequisites

​Step 1: Find the Expert Identifier

​Step 2: Single-Turn Invocation

​Step 3: Multi-Turn Conversations

​Step 4: Handling the SSE Stream

​Step 5: Error Handling

​Step 6: Building a Chat Interface

​Complete Example

​Expert Invocation vs AI Model Calls

​Next Steps

SDK Quick Start

AI Model Calls

App Integration Tutorial

API Reference

What You’ll Build

What Is an Expert

Prerequisites

Step 1: Find the Expert Identifier

Step 2: Single-Turn Invocation

Step 3: Multi-Turn Conversations

Step 4: Handling the SSE Stream

Step 5: Error Handling

Step 6: Building a Chat Interface

Complete Example

Expert Invocation vs AI Model Calls

Next Steps