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

# Python SDK

> Profy Python SDK installation and usage guide

`profy` is the official Python SDK for the Profy platform, providing both async and sync clients with built-in authentication, streaming, and error handling.

## Installation

```bash theme={null}
pip install profy
```

Requires Python 3.8+. The SDK is built on `httpx`.

## Quick Start

### Async Client (Recommended)

```python theme={null}
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())
```

### Sync Client

```python theme={null}
from profy import ProfySync

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

## Client Configuration

```python theme={null}
from profy import Profy
import httpx

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

| Parameter  | Type               | Default                | Description                   |
| ---------- | ------------------ | ---------------------- | ----------------------------- |
| `api_key`  | `str`              | (required)             | API Key or OAuth Access Token |
| `base_url` | `str`              | `https://api.profy.cn` | API base URL                  |
| `timeout`  | `Timeout \| float` | 300 seconds            | Request timeout               |

## Resource Namespaces

The SDK organizes API methods through resource namespaces:

| Namespace                 | Methods                 | Endpoint                    |
| ------------------------- | ----------------------- | --------------------------- |
| `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()` — Invoke an Expert

```python theme={null}
result = await client.agents.run(
    "my-expert",           # Expert identifier
    "帮我分析这段代码",      # Message content
    session_id="abc123",   # Optional: continue a session
    model="deepseek-v3",   # Optional: specify a model
)
```

Returns an `AgentRunResponse` object:

| Property      | Type             | Description                                       |
| ------------- | ---------------- | ------------------------------------------------- |
| `text`        | `str`            | The Agent's complete text reply                   |
| `model`       | `str \| None`    | Model name used                                   |
| `session_id`  | `str \| None`    | Session ID (for multi-turn conversations)         |
| `tool_calls`  | `list[ToolCall]` | List of tool calls                                |
| `usage`       | `TokenUsage`     | Token usage                                       |
| `error`       | `str \| None`    | Error message (if any)                            |
| `partial`     | `bool`           | Whether the response was truncated due to timeout |
| `elapsed_sec` | `float`          | Elapsed time (seconds)                            |

### `agents.run_stream()` — Stream an Expert Invocation

```python theme={null}
async for chunk in client.agents.run_stream("my-expert", "你好"):
    if chunk.type == "output.text.delta":
        print(chunk.delta, end="", flush=True)
    elif chunk.type == "tool_call.created":
        print(f"\n🔧 {chunk.raw.get('tool_name')}")
    elif chunk.type == "run.completed":
        print("\nDone")
```

Each `AgentRunChunk` contains:

| Property | Type          | Description                                   |
| -------- | ------------- | --------------------------------------------- |
| `type`   | `str`         | Event type (e.g., `output.text.delta`)        |
| `delta`  | `str \| None` | Text increment (only for `output.text.delta`) |
| `raw`    | `dict`        | Raw event data                                |

## Chat Completions

### `chat.completions.create()` — Chat Completion

**Non-streaming**:

```python theme={null}
resp = await client.chat.completions.create(
    model="deepseek-v3",
    messages=[
        {"role": "system", "content": "You are a helpful assistant"},
        {"role": "user", "content": "你好"},
    ],
    temperature=0.7,
    max_tokens=1000,
)
print(resp.text)
print(f"Tokens: {resp.usage.total_tokens}")
```

**Streaming**:

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

`ChatResponse` object:

| Property | Type         | Description       |
| -------- | ------------ | ----------------- |
| `text`   | `str`        | Reply text        |
| `model`  | `str`        | Model name        |
| `usage`  | `TokenUsage` | Token usage       |
| `raw`    | `dict`       | Raw response data |

## Models

### `models.list()` — Get Model List

```python theme={null}
models = await client.models.list()

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

## Events

### `events.report()` — Report a Custom Event

```python theme={null}
result = await client.events.report(
    "image_generation",
    idempotency_key="order_123_gen_1",
    metadata={"style": "watercolor"},
)
print(f"Charged: {result['charged']}")
print(f"Remaining: {result['balance_remaining']}")
```

| Parameter         | Type   | Required | Description                                 |
| ----------------- | ------ | -------- | ------------------------------------------- |
| `event_name`      | `str`  | ✅        | Event name                                  |
| `idempotency_key` | `str`  | No       | Idempotency key (auto-generated if omitted) |
| `metadata`        | `dict` | No       | Additional metadata                         |

<Warning>
  The Events API only supports OAuth Tokens. Calling with an API Key will return a 401 error.
</Warning>

## Error Handling

The SDK raises `ProfyApiError` when the API returns a 4xx/5xx status code:

```python theme={null}
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 Status: {e.status_code}")
        print(f"Error: {e.body}")
        if e.status_code == 404:
            print("Expert not found")
        elif e.status_code == 403:
            print("Access denied")
        elif e.status_code == 429:
            print("Rate limited, please retry later")
```

### Timeout Handling

```python theme={null}
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("Request timed out")
```

`agents.run()` will attempt to return partial results already received on timeout (`result.partial = True`) rather than raising an exception directly.

## Sync Client

`ProfySync` provides the exact same API but with synchronous methods:

```python theme={null}
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.delta, end="")

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

    # Models
    models = client.models.list()
```

<Note>
  `ProfySync` does not include the `events` namespace, since the Events API requires an OAuth Token and is typically used in async web applications.
</Note>

## Full Example: Multi-Turn Chat Bot

```python theme={null}
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("Start chatting (type 'quit' to exit)\n")

        while True:
            user_input = input("You: ")
            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.raw.get("session_id")
                    elif chunk.type == "output.text.delta":
                        print(chunk.delta, end="", flush=True)
                print()
            except ProfyApiError as e:
                print(f"\nError: {e.body}")

asyncio.run(chat_bot())
```

## Next Steps

<CardGroup cols={2}>
  <Card title="TypeScript SDK" icon="js" href="/en/developers/sdk-typescript">
    TypeScript SDK guide
  </Card>

  <Card title="API Reference" icon="code" href="/en/developers/api/agents-run">
    Complete API documentation
  </Card>
</CardGroup>