plungeai (Python)

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

pip install plungeai

Usage

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 TypedDicts (plungeai.types); responses are plain dicts typed the same way. Path parameters come first, positionally: client.workflows.versions.restore(workflow_id, version).

Async client

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

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

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

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)

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:

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

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.