worldtree-sdk (1.0.0)
Installation
pip install --index-url https://gitea.phasefinal.com/api/packages/vh/pypi/simple/ --extra-index-url https://pypi.org/simple worldtree-sdkAbout this package
Official async Python consumer SDK for the Worldtree Conversation API
worldtree-sdk (Python)
Official async Python consumer SDK for the Worldtree Conversation API — the
Python sibling of the shipped TypeScript v1.0.0, built off the same behavioral
spec and the same fixture corpus (../docs/behavioral-spec.md, ../fixtures/),
re-idiomatized to async Python per the contract at
../docs/contracts/worldtree-sdk-python-v0.contract.md.
- Async-first.
async with WorldtreeClient(...),async for ev in client.sessions.stream_turn(...); cancellation via native asyncio. - Resilient streaming.
stream_turnauto-resumes across connection drops with jittered backoff, replaying only unseen events — one continuous event sequence. - One runtime dependency:
httpx. Stdlibdataclassesevents; no Pydantic. - Wire-verbatim
snake_case. No field-mapping layer; reads are open-world, so additive server fields are never dropped. - Python ≥ 3.11,
mypy --strict-clean, shipspy.typed.
Install
uv pip install worldtree-sdk # from the internal Gitea package index
Quickstart
import asyncio
from worldtree_sdk import WorldtreeClient
async def main() -> None:
async with WorldtreeClient("https://api.worldtree.example", auth="wt-...") as client:
async for ev in client.sessions.stream_turn(session_id, "Hello there"):
if ev.type == "text":
print(ev.content, end="", flush=True)
elif ev.type == "done":
print(f"\n[{ev.phase}] {ev.usage}")
asyncio.run(main())
Streaming a turn
client.sessions.stream_turn(session_id, content) is the default resilient
stream (auto-resume on drops). stream_turn_raw(...) is the single-attempt
primitive if you want to drive reconnection yourself.
Events are frozen dataclasses in a discriminated union (TurnEvent) — switch on
ev.type. Every event carries sse_id, turn_id, and raw (the full parsed
envelope, so unknown/additive fields are always reachable):
ev.type |
dataclass | key fields |
|---|---|---|
text / thinking |
TextEvent / ThinkingEvent |
content |
text_boundary |
TextBoundaryEvent |
kind, char_offset |
tool_start / tool_result |
ToolStartEvent / ToolResultEvent |
name, arguments / result |
affect_update |
AffectUpdateEvent |
status, snapshot |
worker_phase / awaiting_llm_first_token |
… | phase / elapsed_ms_since_building_prompt |
done / error / cancelled |
terminal | response/usage · message/error_code · reason |
Cancel a running turn (independent of the stream, which keeps yielding until its terminal):
await client.sessions.cancel_turn(session_id, turn_id, persist_partial=True)
Abandoning a stream: breaking the async for (or the surrounding task being
cancelled) closes the connection but does not cancel the server turn by default
(abandon="detach"). Pass abandon="cancel" for break-out-means-cancel UX:
async for ev in client.sessions.stream_turn(sid, content, abandon="cancel"):
if ev.type == "text" and stop_requested:
break # closes the connection AND issues a best-effort cancel
REST surface
Every REST call is await-ed and returns the parsed body (open-world dict, or a
list for the array endpoints); 204 responses return None.
# sessions
page = await client.sessions.list(limit=25, include_archived=True)
info = await client.sessions.get(session_id)
msgs = await client.sessions.messages(session_id, limit=50, cursor=page["next_cursor"])
new = await client.sessions.create({"agent_id": "a1", "config": {"system_prompt": "…"}})
await client.sessions.update(session_id, {"name": "Renamed", "archived": True})
await client.sessions.delete(session_id)
# persona state (W-7) — the canonical PAD wrapper; a non-finite axis is rejected pre-HTTP
from worldtree_sdk import PadState
await client.sessions.set_persona_state(session_id, PadState(0.4, -0.1, 0.2))
# authored history (#347) — idempotency_key is REQUIRED and never auto-generated;
# a replay with the same (session, key) returns the original ack.
turn = await client.sessions.write_history(
session_id,
{"author": "assistant", "content": "…", "idempotency_key": "unique-key-1"},
)
# identity / catalogs
me = await client.me.get()
usage = await client.me.usage(window="7d", breakdown=True)
caps = await client.capabilities.get()
models = await client.models.available_for_characters()
# characters
char = await client.characters.create({"character": {"name": "Nyx"}})
state = await client.characters.state(character_id)
# consumer agents (Tier-3)
agents = await client.agents.list()
agent = await client.agents.define(
{"agent_name": "Nyx", "role": "gpt-echo", "system_prompt": "be nyx"}
)
Admin
Admin calls need a separate admin credential and use it per-request — they
never fall back to the default bearer. A client built without admin_auth raises
ConfigurationError before any network I/O.
async with WorldtreeClient(base, auth="user-key", admin_auth="admin-key") as client:
keys = await client.admin.keys.list(user_prefix="acme/")
await client.admin.keys.issue({"label": "ci", "user_id": "u1", "tier": "pro"})
await client.admin.keys.bulk_rotate({"user_prefix": "acme/", "grace_seconds": 300})
await client.admin.persona.archive({"end_user_id": "u1"})
await client.admin.sessions.retire(session_id)
await client.admin.users.change_tier("u1", {"tier": "pro"})
rollup = await client.admin.usage(window="30d", breakdown=True)
async for ev in client.admin.stream_events(): # D2 admin lifecycle stream
...
Auth & transport
-
authis a static key or a provider callable (sync or async) returning a short-lived token — resolved per request:WorldtreeClient(base, auth=lambda: mint_heimdall_token()) # sync provider WorldtreeClient(base, auth=async_token_provider) # async provider -
Bifrost-bound
sessions.createrequires a per-requestconsumer_key(checked pre-HTTP):await client.sessions.create(body, consumer_key="ck-…"). -
Transport ownership. By default the client owns a private
httpx.AsyncClient(closed byaclose()/async with). Inject your own to share a pool — it stays caller-owned and is never closed by the SDK:async with httpx.AsyncClient() as pool: client = WorldtreeClient(base, auth="…", transport=pool) # pool NOT closed by aclose()
Errors
One base, three branches (WorldtreeError): stream/connection failures
(ConnectFailed + AgentNotAvailable / TurnLaunchUnavailable / SessionRetired,
ConnectionDropped, ProtocolErrors, ResumeErrors), CancelErrors, and the REST
ApiError floor. Client-misuse is a plain ValueError/TypeError raised pre-HTTP.
from worldtree_sdk import ApiError
try:
await client.sessions.get(session_id)
except ApiError as e:
print(e.status, e.error_code, e.message) # e.g. 404 "session_not_found"
error_code is an open string — a novel server code is carried verbatim, never
rejected. Hide-existence 404s (feature-absent / ungranted / session-absent) are
surfaced undiscriminated on the ApiError floor by design (no NotFoundError).
Bearer tokens are redacted from all error strings.
Development
uv venv
uv pip install -e '.[dev]'
uv run pytest # conformance (shared ../fixtures corpus) + unit tests
uv run mypy # strict typing gate
uv run ruff check . # lint gate
The conformance runner exercises the shared ../fixtures/*.json corpus — the
executable acceptance floor. The corpus is the floor, not the ceiling: every B-*
behavioral rule binds whether or not a fixture covers it.