Official Python SDK for the Hot Dev API.
- PyPI distribution:
hot-dev - Import package:
hot - Tested on Python 3.10, 3.11, 3.12, 3.13
pip install hot-devimportosfromhotimportHotClienthot=HotClient(token=os.environ["HOT_API_KEY"])
# base_url defaults to https://api.hot.dev.# For local development with `hot dev`, pass base_url="http://localhost:4681".foreventinhot.streams.subscribe_with_event(
{
"event_type": "team-agent:ask",
"event_data": {
"session_id": "web:chat:demo",
"user_id": "web:user:demo",
"user_name": "Demo User",
"question": "what is blocking launch?",
},
}
):
ifevent["type"] =="stream:data":
print(event["data_type"], event.get("payload"))
ifevent["type"] =="run:stop":
print(event.get("run", {}).get("result"))
breakThe client closes its underlying HTTP connection cleanly when used as a context manager:
withHotClient(token=os.environ["HOT_API_KEY"]) ashot:
event=hot.events.publish(
{
"event_type": "team-agent:ask",
"event_data": {"question": "what changed?"},
}
)
print(event["stream_id"])Authenticated clients should run server-side. Browser apps and public notebooks should call your own backend route instead of exposing a Hot API key. Most management endpoints require an API key. Sessions and service keys are permission-scoped and are mainly for event publishing and stream reads.
When code has a run id but does not need the full stream, wait on its durable terminal snapshot:
run=hot.runs.wait(run_id, timeout=300)
print(run["result"])When a run returns a background task id, wait through the durable task resource:
task=hot.tasks.wait(task_id, timeout=300)
print(task["result"])The task waiter receives the latest persisted state first and reconnects, so it
cannot miss a task that completed before subscription. A failed, cancelled, or
timed-out task raises HotTaskError with the final task record.
The task's parent stream also emits durable task:update events, which is
useful when one subscription needs to coordinate several tasks.
importosfromhotimportAsyncHotClientasyncwithAsyncHotClient(token=os.environ["HOT_API_KEY"]) ashot:
event=awaithot.events.publish(
{
"event_type": "team-agent:ask",
"event_data": {"question": "what changed?"},
}
)
print(event["stream_id"])Async streaming is supported the same way:
asyncwithAsyncHotClient(token=os.environ["HOT_API_KEY"]) ashot:
asyncforeventinhot.streams.subscribe_with_event(
{
"event_type": "team-agent:ask",
"event_data": {"question": "what changed?"},
}
):
print(event["type"], event)AsyncHotClient exposes the same resource namespaces as HotClient.
Non-2xx API responses raise HotApiError with structured fields:
fromhotimportHotApiErrortry:
hot.projects.get("missing-project")
exceptHotApiErroraserror:
print(error.status_code, error.code, error.request_id, error.retry_after)JSON requests are retried automatically (at most twice) when the API responds
429 with a retry_after; other errors are raised as-is. Streaming and raw
requests are never retried.
HotClient mirrors the Hot API v1 resources:
hot.events— publish, list, get, inspect event runs, and call Hot functions withcall_hot(fn, args)hot.streams— subscribe to run and task updates, wait for run results, and publish events atomically (reconnects automatically across the 5-minute SSE timeout; passreconnect=Falseto opt out)hot.runs— list, inspect, subscribe to, and wait for durable runshot.tasks— get, subscribe to, and wait for durable background taskshot.files— upload, download, list, and delete files (including multipart uploads)hot.projects— create, list, update, activate, deactivate, and delete projectshot.builds— upload, download, deploy, and look up live/deployed buildshot.context— manage encrypted project context variableshot.domains— register, verify, list, and delete custom domainshot.sessions— create and revoke scoped sessionshot.service_keys— create and revoke scoped service keyshot.org— view usage and limitshot.env— read environment info and subscribe to environment events
Use hot.request(...) or hot.request_raw(...) as an escape hatch for API
endpoints that do not yet have a resource helper.
hot.env.subscribe() requires API key credentials and a live API pub/sub
backend; local API servers without pub/sub return a 503. subscribe_with_event
reconnects across the API's 5-minute SSE timeout and stops after the terminal
run correlated to the event it published. Unrelated runs on the same stream do
not end the iterator.
HotClient and AsyncHotClient accept an httpx.Client (or
httpx.AsyncClient) via the client= argument. Use this to configure retries,
proxies, or custom transports:
importhttpxfromhotimportHotClienthttp_client=httpx.Client(
timeout=httpx.Timeout(connect=5.0, read=60.0, write=30.0, pool=5.0),
transport=httpx.HTTPTransport(retries=3),
)
hot=HotClient(token=os.environ["HOT_API_KEY"], client=http_client)When you pass your own client, the SDK does not close it on exit.
Core API request and response payloads use the Hot API wire format:
event_type, event_data, stream_id, and so on. SDK-only options use Python
style names such as base_url and timeout.
The SDK never transforms user-owned payloads such as event_data.
The package ships with py.typed, so editors and type checkers will pick up
the public API. Request and response payloads are currently typed as plain
dict[str, Any] — keys follow the Hot API wire format documented above. Strict
TypedDict shapes generated from the Hot API OpenAPI spec are planned for a
future release.
python -m venv .venv
source .venv/bin/activate
python -m pip install -e ".[dev]"
ruff check .
pytest
python -m buildApache-2.0