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

# plungeai (Python)

> The official Python SDK for the One API — sync and async clients, typed resources, streaming, pagination, retries and errors, generated from the OpenAPI document.


The official Python SDK for the PlungeAI One API (`https://api.plungeai.com`). Targets One API `2.4.0`; package version `1.0.0`. The resource methods and their types are generated from the API's own OpenAPI document (`GET /v1/openapi.json`). One runtime dependency: `httpx`.

Release status: `pip index versions plungeai`; until it answers, build from `sdk/one-api-python/` (`bash scripts/build.sh`, then `pip install target/sdk-py-build/*.whl`).

## Installation

```sh
pip install plungeai
```

## Usage

```python
from plungeai import PlungeAI

client = PlungeAI()  # reads PLUNGEAI_API_KEY; or PlungeAI(api_key="ozk_…")

me = client.whoami()
run = client.agents.execute("calculator-agent", prompt="6*7")
hits = client.discovery.search(q="web search", kind="agents")
```

Method names are `snake_case`; request fields are keyword arguments typed as `TypedDict`s (`plungeai.types`); responses are plain `dict`s typed the same way. Path parameters come first, positionally: `client.workflows.versions.restore(workflow_id, version)`.

## Async client

```python
import asyncio
from plungeai import AsyncPlungeAI

async def main() -> None:
    client = AsyncPlungeAI()
    run = await client.agents.execute("calculator-agent", prompt="6*7")
    async for event in await client.workflows.execute_inline_stream(workflow=workflow):
        print(event.event, event.data)

asyncio.run(main())
```

Every method exists on both clients with the same signature.

## Streaming responses

```python
stream = client.workflows.execute_inline_stream(workflow=workflow)
for event in stream:
    if event.event == "token":
        print(event.json()["delta"], end="")
```

`event.data` is the raw string of the server-sent event; `event.json()` parses it. Leaving the loop closes the underlying response. Methods that take `stream=True` (`client.agents.execute`, `client.chat.completions.create`) return the same `Stream`.

## Auto-pagination

```python
for w in client.workflows.list(kind="bot"):
    print(w)

page = client.workflows.list(limit=50)
for w in page.data:
    print(w)
while page.has_next_page():
    page = page.get_next_page()
    for w in page.data:
        print(w)
```

## Handling errors

```python
from plungeai import APIError, FailedDependencyError

try:
    client.agents.execute("calculator-agent", prompt="6*7")
except FailedDependencyError as err:
    print(err.connect["link"])
except APIError as err:
    print(err.status_code, err.code, err.request_id)
```

| Status | Error class |
|---|---|
| 400 | `BadRequestError` |
| 401 | `AuthenticationError` |
| 403 | `PermissionDeniedError` |
| 404 | `NotFoundError` |
| 409 | `ConflictError` |
| 422 | `UnprocessableEntityError` |
| 424 | `FailedDependencyError` — `connect: { provider, method, link }` says where to connect the missing account |
| 429 | `RateLimitError` |
| ≥ 500 | `InternalServerError` |
| no response | `APIConnectionError` (`APITimeoutError` on timeout) |

Every class is importable from `plungeai`. `err.code` is the envelope's `error.code`; `err.request_id` is the `x-request-id` header.

## Retries

Failed requests are retried 2 times by default with short exponential backoff (honouring `Retry-After`): connection errors, timeouts, 408, 429 and ≥ 500. 409 and 424 are never retried — on the One API they are semantic answers (`approval_required`, `duplicate_execution_id`, a missing connection), and a retried execute would start a second run. Set `max_retries` on the client or per call:

```python
client = PlungeAI(max_retries=0)
client.agents.execute("calculator-agent", prompt="6*7", max_retries=5)
```

`idempotency_key` (per call) is sent as `x-trace-id`, the API's idempotency key: it becomes the run's execution id, is re-sent on every retry of the call, and a replay answers 409 `duplicate_execution_id`.

## Timeouts

Requests time out after 10 minutes by default and raise `APITimeoutError` (retried as above). Set `timeout` in seconds on the client or per call: `PlungeAI(timeout=20.0)`, `client.whoami(timeout=5.0)`.

## Raw responses

```python
response = client.agents.with_raw_response.execute("calculator-agent", prompt="6*7")
print(response.status_code, response.headers["x-request-id"])
run = response.parse()
```

`extra_headers` and `extra_query` on any call add to the request; `client.get(path, query=…)`, `client.post(path, body=…)` and friends reach routes the generated layer does not cover; `executions.continue` is spelled `client.executions.continue_(...)` (Python keyword).

## Models plane

`client.chat.completions.create(...)` (plus `client.embeddings` and `client.models`) covers the OpenAI-compatible routes here; the OpenAI SDK also works with `base_url="https://api.plungeai.com/v1"` and the same key.

## Requirements

Python 3.10 or newer. Type-checked with `mypy --strict`; the package ships `py.typed`.