STATEK Quickstart

Install STATEK, define a supervised agent, start a worker with start_statek(...), and submit work through the client-facing API surface.

Install

Create a Python environment and install STATEK with dbzero support:

pip install "statek[dbzero]"

For deployments that need the advanced dbzero-pro storage and security features, install STATEK with the dbzero-pro extra instead:

pip install "statek[dbzero-pro]"

Configure

Initialize dbzero in the process that creates agents, queues, or jobs. Configure at least one model provider for jobs that call an LLM.

mkdir -p ./statek-data
 
export OPENROUTER_API_URL="https://openrouter.ai/api/v1"
export OPENROUTER_API_KEY="..."
export STATEK_LOG_LEVEL=ERROR

Open a dbzero root directory and prefix before constructing persisted STATEK objects:

import dbzero as db0
 
DBZERO_ROOT = "./statek-data"
PREFIX = "/acme/triage/dev/jobs"
 
 
def open_statek_store():
    db0.init(dbzero_root=DBZERO_ROOT, read_write=True)
    db0.open(PREFIX, "rw")

See Model Providers for provider keys, model strings, and custom provider setup.

Define an agent

A SupervisedAgent can be started by application code, queues, or the client API. The MODEL metadata is required when STATEK creates jobs for the agent.

In production applications, system prompts and model metadata are usually loaded from .md prompt definition files. This quickstart uses the code-only form so you can see the agent internals directly and learn the Python API surface before moving those definitions into files. See Prompt Definitions for the production file format.

from statek.agents.agent import SupervisedAgent
from statek.prompt_config import make_system_prompt
 
 
def create_triage_agent():
    agent = SupervisedAgent(
        role="triage_agent",
        _system_prompt=make_system_prompt(
            "You are a concise triage agent. Inspect the event and decide the next action."
        ),
        _tools=[],
        _metadata={"MODEL": "openrouter/openai/gpt-5-mini"},
    )
 
    agent.update_warmup_def(
        """
print("received event:", event)
"""
    )
    return agent

The warmup code references one external variable, event. That is the variable STATEK can bind when you create a job with shared_vars={"event": ...}.

Start the worker

Create a StatekPushQueue on the prefix where STATEK should look for push notifications to existing jobs, then start the blocking worker entrypoint:

from statek import start_statek
from statek.statek_push_queue import StatekPushQueue
 
from quickstart_common import PREFIX, create_triage_agent, open_statek_store
 
 
open_statek_store()
 
agent = create_triage_agent()
queue = StatekPushQueue(prefix=PREFIX)
 
start_statek(
    agents=[agent],
    push_queues=[queue],
    max_concurrency=10,
)

start_statek(...) prepares STATEK, creates the StatekClientAPI singleton in the worker process, registers the push-notification prefixes from push_queues, and runs the worker loop until the loop returns or the process stops. push_queues is not the application task queue; your application still owns task selection, locking, retries, and prioritization.

For services that already own an event loop, use the async entrypoint:

from statek import start_statek_async
from statek.statek_push_queue import StatekPushQueue
 
from quickstart_common import PREFIX, create_triage_agent, open_statek_store
 
 
async def run_worker():
    open_statek_store()
 
    agent = create_triage_agent()
    queue = StatekPushQueue(prefix=PREFIX)
 
    await start_statek_async(
        agents=[agent],
        push_queues=[queue],
        max_concurrency=10,
    )

Submit work

There are two simple ways to submit work to STATEK:

• Create a job directly with StatekClientAPI.
• Put work in an application-owned task queue and let an agent's warmup code pick the next task with an application-owned temporal function.

Use StatekClientAPI().submit_new_job(...) when a host service, script, webhook handler, or RPC endpoint already knows which agent should handle the work. The host can expose this path through db0-rpc, HTTP, a CLI, or any other interface it owns.

from statek.statek_client_api import StatekClientAPI
 
from quickstart_common import create_triage_agent, open_statek_store
 
 
open_statek_store()
 
agent = create_triage_agent()
incoming_event = {
    "type": "support_ticket",
    "subject": "Invoice total looks wrong",
    "priority": "normal",
}
 
job = StatekClientAPI().submit_new_job(
    agent=agent,
    shared_vars={"event": incoming_event},
    locale="EN-US",
    source="quickstart",
)
 
print("submitted job:", job)

Use queue-backed warmup when the application wants STATEK jobs to behave like durable workers. In this shape, submitting work means adding a task to your application's queue. The queue, locking, retries, task selection, and temporal function are all application code.

The temporal helper checks the queue. If a task is ready, warmup receives it immediately. If no task is ready, the job can suspend and later resume when the temporal condition becomes true.

from statek.exceptions import FutureError
from statek.future import FutureResult, temporal
 
 
def fetch_task(future):
    task_queue = future.deps
    task = task_queue.peek_ready_task()
    if task is None:
        raise FutureError(future)
    return task
 
 
def task_available(future):
    return future.deps.has_ready_task()
 
 
@temporal(complement=fetch_task, condition=task_available)
def pick_next_task(task_queue):
    task = task_queue.peek_ready_task()
    if task is not None:
        return task
    return FutureResult(deps=task_queue, state_num=task_queue.version)

Then the agent's warmup code can prepare the job workspace from the next queued task:

task = pick_next_task(task_queue)
event = task.event
 
print("received event:", event)

Use this pattern for bounded worker pools where the queue readiness check is cheap and the application controls how many jobs can wait this way. For broad user-facing event delivery, prefer a direct job or callback/event path.

Production shape

A production-shaped STATEK deployment usually separates three responsibilities:

• A worker process runs start_statek(...) or start_statek_async(...).
• A hosting service opens the relevant dbzero state and may expose db0-rpc at /rpc.
• External clients call RPC-exposed StatekClientAPI methods or host-owned endpoints that enqueue application work.

STATEK provides the StatekClientAPI object, remote-decorated methods, persisted jobs, job notifications, and the built-in restricted Python sandbox. That sandbox is backed by dbzero restricted prefixes for dbzero objects exposed to agent code. For deployments that need stronger data-access controls, dbzero-pro can add protected fields and data filtering predicates for per-user, per-tenant, or per-role access.

STATEK does not start, secure, authenticate, or operate /rpc for you. The host application still owns RPC setup, auth, networking, secrets, tenant policy, process and resource limits, monitoring, deployment, failure handling, and side-effect controls.

Next steps

Read Operations for worker behavior and deployment concerns, Client API for job submission, Runners for start_statek(...), Security before exposing code execution to real users, and Model Providers for provider configuration.