Python SDK

Python SDK 当前处于冻结状态，不建议在生产中使用。PyPI 上的 profy-sdk 停留在 0.1.0，只包含 OAuth 与事件上报（ProfyApp）， 不包含本页描述的 Profy 统一客户端。本页所有示例针对的是尚未发布的 1.0.0 源码，照本页 pip install 后 from profy import Profy 会直接 ImportError。在 Python SDK 恢复发布之前，请直接调用 REST /v1 接口，或使用 TypeScript SDK。

profy 是袋袋平台的官方 Python SDK，提供异步和同步两种客户端，封装了认证、流式解析、错误处理等细节。

安装

pip install profy-sdk

发行包名为 profy-sdk，导入名为 profy（from profy import Profy）。要求 Python 3.9+。SDK 基于 httpx 构建。

快速开始

异步客户端（推荐）

import asyncio
from profy import Profy

async def main():
    async with Profy(api_key="sk-pro-your-key") as client:
        result = await client.agents.run("my-expert", "你好")
        print(result.text)

asyncio.run(main())

同步客户端

from profy import ProfySync

with ProfySync(api_key="sk-pro-your-key") as client:
    result = client.agents.run("my-expert", "你好")
    print(result.text)

客户端配置

from profy import Profy
import httpx

client = Profy(
    api_key="sk-pro-your-key",
    base_url="https://api.profy.cn",  # 默认值
    timeout=httpx.Timeout(300.0, connect=10.0),  # 默认值
)

参数	类型	默认值	说明
`api_key`	`str`	（必填）	API Key 或 OAuth Access Token
`base_url`	`str`	`https://api.profy.cn`	API 基础 URL
`timeout`	`Timeout \| float`	300 秒	请求超时时间

资源命名空间

SDK 通过资源命名空间组织 API 方法：

命名空间	方法	对应端点
`client.agents`	`run()`, `run_stream()`	`POST /v1/agents/run`
`client.chat.completions`	`create()`	`POST /v1/chat/completions`
`client.models`	`list()`	`GET /v1/models`
`client.events`	`report()`	`POST /v1/events`

Agents

`agents.run()` — 调用专家

result = await client.agents.run(
    "my-expert",           # 专家标识符
    "帮我分析这段代码",      # 消息内容
    session_id="abc123",   # 可选：延续会话
    model="deepseek-v3",   # 可选：指定模型
)

返回 AgentRunResponse 对象：

属性	类型	说明
`text`	`str`	Agent 的完整文本回复
`model`	`str \| None`	使用的模型名称
`session_id`	`str \| None`	会话 ID（用于多轮对话）
`tool_calls`	`list[ToolCall]`	工具调用列表
`usage`	`TokenUsage`	Token 用量
`error`	`str \| None`	错误信息（如有）
`partial`	`bool`	是否因超时截断
`elapsed_sec`	`float`	耗时（秒）

`agents.run_stream()` — 流式调用专家

async for chunk in client.agents.run_stream("my-expert", "你好"):
    if chunk.type == "output.text.delta":
        print(chunk.content, end="", flush=True)
    elif chunk.type == "tool_call.created":
        print(f"\n🔧 {chunk.tool_name}")
    elif chunk.type == "run.completed":
        print("\n完成")

每个 AgentRunChunk 包含：

属性	类型	说明
`type`	`str`	事件类型（如 `output.text.delta`）
`content`	`str \| None`	文本增量（仅 `output.text.delta`）
`tool_name`	`str \| None`	工具名称（仅 `tool_call.*`）
`usage`	`TokenUsage \| None`	Token 用量（仅 `usage` 事件）
`raw`	`dict`	原始事件数据

Chat Completions

`chat.completions.create()` — 对话补全

非流式：

resp = await client.chat.completions.create(
    model="deepseek-v3",
    messages=[
        {"role": "system", "content": "你是一个有帮助的助手"},
        {"role": "user", "content": "你好"},
    ],
    temperature=0.7,
    max_tokens=1000,
)
print(resp.text)
print(f"Token: {resp.usage.total_tokens}")

流式：

stream = await client.chat.completions.create(
    model="deepseek-v3",
    messages=[{"role": "user", "content": "讲一个故事"}],
    stream=True,
)
async for chunk in stream:
    if chunk.delta.content:
        print(chunk.delta.content, end="", flush=True)

ChatResponse 对象：

属性	类型	说明
`text`	`str`	回复文本（`choices[0].message.content` 的快捷属性）
`model`	`str`	模型名称
`usage`	`TokenUsage`	Token 用量
`id`	`str`	响应 ID
`created`	`int`	创建时间戳
`choices`	`list[ChatChoice]`	原始 choices 列表

Models

`models.list()` — 获取模型列表

models = await client.models.list()

for m in models:
    print(f"{m['id']}: {m['name']}")
    caps = m.get("capabilities", {})
    print(f"  工具: {'✅' if caps.get('supportsTools') else '❌'}")
    print(f"  视觉: {'✅' if caps.get('supportsVision') else '❌'}")

Events

`events.report()` — 上报自定义事件

result = await client.events.report(
    "image_generation",
    idempotency_key="order_123_gen_1",
    metadata={"style": "watercolor"},
)
print(f"扣费: {result['charged']}")
print(f"剩余: {result['balance_remaining']}")

参数	类型	必填	说明
`event_name`	`str`	✅	事件名称
`idempotency_key`	`str`	否	幂等键（不传则自动生成）
`metadata`	`dict`	否	附加元数据

/v1/events 需要 OAuth Access Token（把它作为 api_key 传入客户端）。使用 sk-pro- API Key 调用会返回 401（Custom events require OAuth token, not API key）。

错误处理

SDK 会在 API 返回 4xx/5xx 状态码时抛出 ProfyApiError：

from profy import Profy, ProfyApiError

async with Profy(api_key="sk-pro-your-key") as client:
    try:
        result = await client.agents.run("non-existent", "你好")
    except ProfyApiError as e:
        print(f"HTTP 状态码: {e.status_code}")
        print(f"错误信息: {e.body}")
        if e.status_code == 404:
            print("专家 不存在")
        elif e.status_code == 403:
            print("访问被拒绝")
        elif e.status_code == 429:
            print("速率限制，请稍后重试")

超时处理

import httpx
from profy import Profy

async with Profy(
    api_key="sk-pro-your-key",
    timeout=httpx.Timeout(60.0, connect=5.0),
) as client:
    try:
        result = await client.agents.run("my-expert", "复杂任务")
    except httpx.TimeoutException:
        print("请求超时")

agents.run() 在超时时会尽量返回已收到的部分结果（result.partial = True），而不是直接抛出异常。

同步客户端

ProfySync 提供完全相同的 API，但使用同步方法：

from profy import ProfySync

with ProfySync(api_key="sk-pro-your-key") as client:
    # Agents
    result = client.agents.run("my-expert", "你好")
    for chunk in client.agents.run_stream("my-expert", "你好"):
        print(chunk.content, end="")

    # Chat
    resp = client.chat.completions.create(
        model="deepseek-v3",
        messages=[{"role": "user", "content": "你好"}],
    )

    # Models
    models = client.models.list()

ProfySync 与 Profy 提供完全一致的资源命名空间（含 events）。注意：events.report() 需用 OAuth Access Token 构造客户端（ProfySync(api_key=oauth_token)）；用 sk-pro- API Key 调用 /v1/events 会返回 401。

完整示例：多轮对话 Bot

import asyncio
from profy import Profy, ProfyApiError

async def chat_bot():
    async with Profy(api_key="sk-pro-your-key") as client:
        session_id = None
        print("开始对话（输入 'quit' 退出）\n")

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

            try:
                print("AI: ", end="", flush=True)
                async for chunk in client.agents.run_stream(
                    "my-expert",
                    user_input,
                    session_id=session_id,
                ):
                    if chunk.type == "run.in_progress":
                        session_id = chunk.session_id
                    elif chunk.type == "output.text.delta":
                        print(chunk.content, end="", flush=True)
                print()
            except ProfyApiError as e:
                print(f"\n错误: {e.body}")

asyncio.run(chat_bot())

开始使用

认证与密钥

API 能力指南

SDK

集成

Cookbook

参考

安装

快速开始

异步客户端（推荐）

同步客户端

客户端配置

资源命名空间

Agents

`agents.run()` — 调用专家

`agents.run_stream()` — 流式调用专家

Chat Completions

`chat.completions.create()` — 对话补全

Models

`models.list()` — 获取模型列表

Events

`events.report()` — 上报自定义事件

错误处理

超时处理

同步客户端

完整示例：多轮对话 Bot

下一步

TypeScript SDK

API 参考

​安装

​快速开始

​异步客户端（推荐）

​同步客户端

​客户端配置

​资源命名空间

​Agents

​agents.run() — 调用专家

​agents.run_stream() — 流式调用专家

​Chat Completions

​chat.completions.create() — 对话补全

​Models

​models.list() — 获取模型列表

​Events

​events.report() — 上报自定义事件

​错误处理

​超时处理

​同步客户端

​完整示例：多轮对话 Bot

​下一步

TypeScript SDK

API 参考

安装

快速开始

异步客户端（推荐）

同步客户端

客户端配置

资源命名空间

Agents

`agents.run()` — 调用专家

`agents.run_stream()` — 流式调用专家

Chat Completions

`chat.completions.create()` — 对话补全

Models

`models.list()` — 获取模型列表

Events

`events.report()` — 上报自定义事件

错误处理

超时处理

同步客户端

完整示例：多轮对话 Bot

下一步