coze-coding-utils

Python utilities for the Coze coding client. Provides a simple runtime Context object and an async-task subsystem.

Installation

pip install coze-coding-utils

Modules

runtime_ctx

from coze_coding_utils.runtime_ctx.context import Context

ctx = Context(run_id="r", space_id="s", project_id="p", method="fetchData")
print(ctx.run_id, ctx.method)

async_tasks

Per-process async task runtime backed by PostgreSQL (single table, JSONB payload, partial indexes). Public API:

from coze_coding_utils.async_tasks import (
    AsyncTaskRuntime,        # construct one per process
    AsyncTask,               # SQLAlchemy ORM (single PK on task_id)
    AsyncTaskStatus,         # pending / running / completed / failed / timeout
    AsyncTaskBase,           # DeclarativeBase if caller needs to drive DDL
    AsyncTaskStorageError,   # raised on DB init/query failure (caller turns into 503)
    config,                  # env-overridable constants module
)

runtime = AsyncTaskRuntime(
    session_factory=get_session,   # Callable[[], sqlalchemy.orm.Session]
    engine=get_engine(),           # sqlalchemy.engine.Engine
    graph=async_graph,             # any object with .ainvoke(payload, config=...)
    checkpointer=checkpointer,     # any LangGraph-compatible checkpointer
)

# Lifespan / FastAPI usage:
async with lifespan(...):
    await runtime.submit(payload={...}, biz_context={"x-user-id": "u1"}, deadline_sec=600)
    row = await runtime.get(task_id)
    ...
    await runtime.shutdown()

The runtime is intentionally framework-agnostic:

• No FastAPI dependency — caller maps AsyncTaskStorageError to HTTP 503 themselves.
• No LangGraph dependency in [project.dependencies] — the runtime only calls graph.ainvoke(payload, config=run_config) and holds the checkpointer reference for resume scenarios.

Caller contracts (pre-conditions)

The runtime makes three assumptions about the caller's environment. Violating any one will silently break checkpoint cleanup or task lifecycle:

1. Checkpointer compatible with PG. Construct the LangGraph with a checkpointer that writes to the schema configured by COZE_ASYNC_TASK_CHECKPOINT_SCHEMA (default memory), tables checkpoints / checkpoint_writes. langgraph-checkpoint-postgres works out of the box.
2. thread_id == task_id. The runtime fixes {"configurable": {"thread_id": task_id}} when calling graph.ainvoke. The memory_cleanup_once scan deletes orphan checkpoint rows by joining on this equality. Don't override thread_id in your graph.
3. Session schema access. The injected session_factory must produce sessions with read+write privileges on COZE_ASYNC_TASK_CHECKPOINT_SCHEMA schema (so cleanup can DELETE orphan rows).

Environment variables

All prefixed with COZE_ASYNC_TASK_. Override only when there's a specific ops reason.

The runtime lazily issues CREATE TABLE IF NOT EXISTS on the first submit / get, so a fresh deployment needs no manual DDL.

Python Version

Requires Python 3.10+.

License

MIT