---
title: "How to build an agent state machine in Pixeltable"
date: "2026-11-03"
dateModified: "2026-11-03"
author: "Pierre Brunelle"
tags:
  - Agent State Machine
  - TableModel
  - Computed Columns
  - MCP
  - Pixeltable Cloud
  - Workflow
description: "The model proposes a transition. A Pixeltable table decides whether it is allowed. This post builds a task queue with guarded updates, computed-column evidence, version history, and HTTP or MCP tools any coding agent can call."
url: "https://www.pixeltable.com/blog/agent-state-machine-pixeltable"
---

# How to build an agent state machine in Pixeltable

**Summary:** You will have a `tasks` table that is the state machine of the work, Python functions that refuse illegal transitions, a computed column that scores a draft before approval can succeed, and HTTP routes (or MCP tools) that Claude Code, Codex, or a custom agent can call. I ran every listing on Pixeltable 0.7.23 and mcp 2.3.0 in a Linux container. The judge variant ran with a stub in place of the model, and I did not call a model. The model never writes `state` on its own.

 
Josh Rosen's article [Agent State Machines: Building Deterministic Scaffolding for Agents](https://x.com/JoshARosen/status/2117687393213932003) (3 November 2026) puts the same split in one sentence. The model proposes a transition. Software decides whether it is allowed. Approval before action, recovery from a failed tool call, and validation before completion are rules. They belong in code, not in a prompt.

 
Rosen names four places a state machine shows up: inside one agent loop, as a workflow graph, as a supervisor of many runs, and as the work itself (his ThruWire product: checkpoints with requirements, evidence, and dependencies). This post is the fourth. If you want a DAG of LLM nodes, that is [The Pipeline Is the Table](/blog/ai-automation-workflow) and the [agentic workflows](https://docs.pixeltable.com/use-cases/agentic-workflows) docs. If you want durable execution of the loop, skip to the last section and use Temporal, Restate, DBOS, or LangGraph. Pixeltable is the wrong tool for that job.

 
## Create the project

 
```bash
pip install -U 'pixeltable[serve]'
pxt init
# app.py holds the Tasks model, the transition functions, and the HTTP routes
pxt schema update app.py agent_sm -f
pxt service update app.py agent_sm -f
pxt service list agent_sm --json
```

 
Python 3.11 or later. `pxt init` marks the project root. `pxt schema update` creates the tables. `pxt service update` starts HTTP. Pass `-f` when a script or a coding agent runs these, because without a terminal they refuse. The service picks its own port. Read the endpoint from `pxt service list --json`. My service answered at `http://127.0.0.1:38611`.

 
## The work is a table

 
One row is one unit of work. The columns are the machine: who owns it, the draft, the reviewer's note, and two computed columns that score that draft.

 
```python
import datetime
from typing import Any
from uuid import UUID

import fastapi
import pixeltable as pxt
import pixeltable.functions as pxtf
from pixeltable.serving import FastAPIRouter
from pydantic import BaseModel

TableModel = pxt.model_base()

QUEUED = 'queued'
CLAIMED = 'claimed'
DRAFTING = 'drafting'
AWAITING_APPROVAL = 'awaiting_approval'
APPROVED = 'approved'
DONE = 'done'
FAILED = 'failed'

@pxt.udf
def check_draft(title: str, brief: str, draft: str | None) -> dict:
 """Rule check that runs when title, brief, or draft changes.

 Swap this function for an LLM judge. The column contract stays a dict
 with ok and reason. See the OpenAI variant below. I ran this stub.
 """
 if draft is None or draft.strip() == '':
 return {'ok': False, 'reason': 'no draft'}
 text = draft.strip()
 if len(text) < 40:
 return {'ok': False, 'reason': 'draft shorter than 40 characters'}
 haystack = text.lower()
 title_hit = title.lower() in haystack
 brief_words = [w for w in brief.lower().split() if len(w) > 3][:4]
 brief_hit = any(w in haystack for w in brief_words)
 if not title_hit and not brief_hit:
 return {'ok': False, 'reason': 'draft does not mention the title or the brief'}
 return {'ok': True, 'reason': 'length and topic checks passed'}

@pxt.udf
def evidence_passed(evidence: dict) -> bool:
 """Boolean gate used in SQL where clauses. Nested JSON is not SQL-filterable."""
 return bool(evidence.get('ok'))

class Tasks(TableModel, name='tasks'):
 id = pxt.Column(value=pxtf.uuid.uuid7(), primary_key=True)
 title: pxt.String
 brief: pxt.String
 state: pxt.String
 owner: pxt.String | None
 draft: pxt.String | None
 review_note: pxt.String | None
 created_at: pxt.Timestamp
 updated_at: pxt.Timestamp
 evidence = check_draft(title, brief, draft)
 passed = evidence_passed(evidence)
```

 
A new row starts `queued`, with `evidence.ok` false because there is no draft yet. That is ThruWire's idea on one table: the work does not complete without evidence. [A computed column](/learn/what-is-a-computed-column) runs when its inputs change. You do not call the judge from the agent loop.

 
## Guard every transition

 
Each event is a function. The function names the states that may accept it, writes the destination, and returns the updated row or `None`. The model asks. The `where` clause answers.

 
```python
def _now() -> datetime.datetime:
 return datetime.datetime.now(datetime.timezone.utc)

def _task_id(task_id: str | UUID) -> UUID:
 return task_id if isinstance(task_id, UUID) else UUID(str(task_id))

def _row(status: pxt.UpdateStatus) -> dict[str, Any] | None:
 """One updated row, or None when the where clause matched nothing."""
 n = status.row_count_stats.upd_rows
 if n == 0:
 return None
 if n != 1:
 raise RuntimeError(f'guarded update matched {n} rows; expected 0 or 1')
 assert status.rows is not None and len(status.rows) == 1
 return status.rows[0]

def enqueue(title: str, brief: str) -> dict[str, Any]:
 ts = _now()
 status = Tasks.insert(
 [
 {
 'title': title,
 'brief': brief,
 'state': QUEUED,
 'owner': None,
 'draft': None,
 'review_note': None,
 'created_at': ts,
 'updated_at': ts,
 }
 ],
 return_rows=True,
 )
 assert status.rows is not None and len(status.rows) == 1
 return status.rows[0]

def claim_task(agent_id: str, task_id: str | None = None) -> dict[str, Any] | None:
 """Claim one queued task. Re-checks state in the update, so a lost race returns None."""
 if task_id is None:
 pending = (
 Tasks.where(Tasks.state == QUEUED)
 .order_by(Tasks.created_at)
 .limit(1)
 .select(Tasks.id)
 .collect()
 )
 if len(pending) == 0:
 return None
 task_id = str(pending[0]['id'])
 return _row(
 Tasks.update(
 {
 'state': CLAIMED,
 'owner': agent_id,
 'updated_at': _now(),
 },
 where=(Tasks.id == _task_id(task_id)) & (Tasks.state == QUEUED),
 return_rows=True,
 )
 )
```

 
Two details that are not obvious. A filter on the UUID primary key must use a `uuid.UUID`. A string raises "Filter not expressible in SQL". And you cannot write `.limit(1).update()`. `Query.update` uses the `where` clause only. Pick one id, then update that id while it is still `queued`.

 
I enqueued "Release notes", then posted `/claim` as `worker-a` and again for the same id as `worker-b`. The first answered 200 and the second 409. Two threads posting `/claim` for one queued id: `racer-0` got 200, `racer-1` got 409.

 
### There is no compare-and-set API

 
Pixeltable does not expose compare-and-set. The pattern that ran is: `update(..., where=..., return_rows=True)` and then `status.row_count_stats.upd_rows`. Zero means the event was illegal or someone else got there first.

 
Default tables are data-versioned. A data write takes an exclusive table lock, so writers on that table serialize. The second claimer waits, then its `where` clause sees `claimed` and matches nothing. That is stronger isolation than two overlapping row updates. It is also a throughput limit: an update of a different row waited for the first writer to finish (the measurement is under the judge listing), so many agents claiming different rows take the table one at a time. If you need a high-rate queue, this table is the wrong tool. Use a durable-execution engine, or a queue product, and keep Pixeltable for the artifacts the agents write.

 
One limit of the code below: `request_approval` is two updates, one for the draft and one for the state. I paused one submission between its two updates, submitted a second draft, and had a reviewer reject it. When the first submission resumed, the row went back to `awaiting_approval` with the rejected draft and the reviewer's note still on it, and nobody had resubmitted. The evidence gate still holds, since `passed` is read from the stored draft. Keep one submitter per task, or serialize submissions in the caller.

 
## Done requires evidence

 
`request_approval` writes the draft first. That recomputes `evidence` and `passed`. A second update moves the row to `awaiting_approval` only when `passed` is true. Nested JSON such as `evidence['ok']` raises "Filter not expressible in SQL" in the `where` of an update. That is why `passed` is a stored bool.

 
```python
def start_draft(task_id: str, agent_id: str) -> dict[str, Any] | None:
 return _row(
 Tasks.update(
 {'state': DRAFTING, 'updated_at': _now()},
 where=(Tasks.id == _task_id(task_id))
 & (Tasks.owner == agent_id)
 & Tasks.state.isin([CLAIMED, DRAFTING]),
 return_rows=True,
 )
 )

def request_approval(task_id: str, agent_id: str, draft: str) -> dict[str, Any] | None:
 """Write the draft, then move to awaiting_approval only if evidence.ok is true.

 None: not this agent's task, or not drafting. A row still in drafting: the check failed.
 """
 wrote = _row(
 Tasks.update(
 {'draft': draft, 'updated_at': _now()},
 where=(Tasks.id == _task_id(task_id))
 & (Tasks.owner == agent_id)
 & (Tasks.state == DRAFTING),
 return_rows=True,
 )
 )
 if wrote is None:
 return None
 promoted = _row(
 Tasks.update(
 {'state': AWAITING_APPROVAL, 'updated_at': _now()},
 where=(Tasks.id == _task_id(task_id))
 & (Tasks.state == DRAFTING)
 & (Tasks.passed == True),
 return_rows=True,
 )
 )
 return promoted or wrote

def complete_task(task_id: str, agent_id: str) -> dict[str, Any] | None:
 """Done requires approved plus evidence.ok. The check is in the where clause."""
 return _row(
 Tasks.update(
 {'state': DONE, 'updated_at': _now()},
 where=(Tasks.id == _task_id(task_id))
 & (Tasks.owner == agent_id)
 & (Tasks.state == APPROVED)
 & (Tasks.passed == True),
 return_rows=True,
 )
 )
```

 
The draft "too short" stayed in `drafting`, and the route answered 409 with `evidence failed: {'ok': False, 'reason': 'draft shorter than 40 characters'}`. A longer draft that named the title moved to `awaiting_approval` with `passed: true`.

 
The run above uses a rule UDF. The column contract does not change if you swap in the [LLM-as-judge](https://docs.pixeltable.com/howto/cookbooks/agents/pattern-llm-judge) pattern. I had no OpenAI key, so I ran this variant with a stub UDF in place of `openai.chat_completions` that returns the shape of the OpenAI response. The stub is a test harness and is not in the listing. No model was called.

 
```python
import json

from pixeltable.functions import openai

JUDGE = """Return JSON with keys ok (boolean) and reason (string).
ok is true only if the draft fulfills the brief.
Title: {title}
Brief: {brief}
Draft: {draft}
"""

@pxt.udf
def judge_messages(title: str, brief: str, draft: str | None) -> list[dict] | None:
 if draft is None:
 return None
 return [
 {'role': 'system', 'content': 'You grade drafts. Reply with JSON only.'},
 {'role': 'user', 'content': JUDGE.format(title=title, brief=brief, draft=draft)},
 ]

@pxt.udf
def parse_judge(raw: dict) -> dict:
 return json.loads(raw['choices'][0]['message']['content'])

# In class Tasks, replace the evidence = check_draft(...) line with these two.
 evidence_raw = openai.chat_completions(messages=judge_messages(title, brief, draft), model='gpt-4o-mini')
 evidence = parse_judge(evidence_raw)
```

 
`judge_messages` returns `None` for a task with no draft, which skips the model call, so a queued task has a null `evidence` and a null `passed`. Building the prompt with `pxtf.string.format(..., draft=draft)` instead raised `KeyError: 'draft'` on 0.7.23 when `draft` was null, which is every new task. With the stub, a draft it accepted gave `{'ok': True, ...}` and `passed` true, and a draft it refused gave `passed` false. Keep `passed = evidence_passed(evidence)` as the gate. Anthropic is `pixeltable.functions.anthropic.messages`, which requires `max_tokens`, with the same messages.

 
The table lock covers the judge call. The stub sleeps 4 seconds where a model would take its latency. One thread updated a draft on it, and an update of a different row started one second later. The second update returned after 3.0 seconds, when the first finished. Check that before you put a slow judge on a busy table. I measured this in one process with two threads, not against a model.

 
## Reject returns the row to drafting

 
```python
def approve(task_id: str, note: str | None = None) -> dict[str, Any] | None:
 return _row(
 Tasks.update(
 {'state': APPROVED, 'review_note': note, 'updated_at': _now()},
 where=(Tasks.id == _task_id(task_id)) & (Tasks.state == AWAITING_APPROVAL),
 return_rows=True,
 )
 )

def reject(task_id: str, note: str) -> dict[str, Any] | None:
 """Rejection returns the row to drafting. The agent keeps ownership."""
 return _row(
 Tasks.update(
 {'state': DRAFTING, 'review_note': note, 'updated_at': _now()},
 where=(Tasks.id == _task_id(task_id)) & (Tasks.state == AWAITING_APPROVAL),
 return_rows=True,
 )
 )

def fail_task(task_id: str, note: str) -> dict[str, Any] | None:
 return _row(
 Tasks.update(
 {'state': FAILED, 'review_note': note, 'updated_at': _now()},
 where=(Tasks.id == _task_id(task_id))
 & Tasks.state.isin([CLAIMED, DRAFTING, AWAITING_APPROVAL, APPROVED]),
 return_rows=True,
 )
 )
```

 
That is Rosen's loop from drafting to approval to sending, with rejection as an edge back to drafting. I rejected "Release notes" with the note "Name the FastAPIRouter change in the first paragraph." The row's `state` became `drafting`, `owner` stayed `worker-a`, and a resubmit went to `awaiting_approval` again. Approve, then complete, left it `done`.

 
`approve`, `reject` and `fail_task` take no caller identity, and this app does not authenticate callers. An agent that can call `/claim` can also call `/approve`: in my run, one caller claimed a task, drafted it, approved it and completed it. Put approval behind a separate credential, or add a `reviewer` argument and a `Tasks.owner != reviewer` predicate. History records each state and `review_note`, not who called.

 
## History is the audit trail

 
Every insert and update is a table version. `history()` is the human report. `get_versions()` is the dict form. Time travel is `pxt.get_table('agent_sm.tasks:N')`, read-only. The docs are [Version control and lineage](https://docs.pixeltable.com/platform/version-control) and the glossary page [What is time travel?](/learn/what-is-time-travel).

 
```python
# REPL, after pxt schema update. Importing app.py does not bind the models.
from app import Tasks, TableModel, DONE, FAILED
import pixeltable as pxt

TableModel.bind_all('agent_sm')
print(Tasks.history(n=8))
v1 = pxt.get_table('agent_sm.tasks:1')
print(v1.select(v1.title, v1.state).collect())

open_rows = (
 Tasks.where((Tasks.state != DONE) & (Tasks.state != FAILED))
 .select(Tasks.id, Tasks.title, Tasks.state, Tasks.owner)
 .order_by(Tasks.updated_at)
 .collect()
)
```

 
After a crash, the same `agent_id` asks `/inbox` and continues its own rows. `start_draft`, `request_approval` and `complete_task` require `owner == agent_id`, so another agent cannot take a dead claimant's row over, and no transition returns a row to `queued`. A release transition would be a new edge. In my run, `crashy` claimed a task, started drafting and "died". `/inbox` for `crashy` listed the row in `drafting`, `/inbox` for `other` did not, and `other` got 409 on `/start-draft` and on `/request-approval`. `crashy` then submitted and the row reached `awaiting_approval`. Nothing in Pixeltable restarts the agent process for you. [Agent memory](/learn/what-is-agent-memory) in Pixeltable is the same idea for conversation state. [Time travel](/blog/pixeltable-versioning-time-travel) is the longer versioning write-up.

 
## Serve claim, approve, and complete over HTTP

 
`FastAPIRouter.add_update_route` updates a row by primary key. A guarded transition needs extra predicates, so the serving docs tell you to use a plain FastAPI `@app.post()` when a helper cannot express the request. Queries still go through `add_query_route`. Put both on one app with `include_router()`.

 
```python
@pxt.query
def open_work() -> pxt.Query:
 return (
 Tasks.where((Tasks.state != DONE) & (Tasks.state != FAILED))
 .select(Tasks.id, Tasks.title, Tasks.state, Tasks.owner, Tasks.review_note)
 .order_by(Tasks.updated_at)
 )

@pxt.query
def inbox(agent_id: str) -> pxt.Query:
 owned = (Tasks.owner == agent_id) & (Tasks.state != DONE) & (Tasks.state != FAILED)
 return (
 Tasks.where((Tasks.state == QUEUED) | owned)
 .select(Tasks.id, Tasks.title, Tasks.brief, Tasks.state, Tasks.review_note)
 .order_by(Tasks.created_at)
 )

@pxt.query
def approval_queue() -> pxt.Query:
 return (
 Tasks.where(Tasks.state == AWAITING_APPROVAL)
 .select(Tasks.id, Tasks.title, Tasks.draft, Tasks.owner, Tasks.evidence)
 .order_by(Tasks.updated_at)
 )

api = FastAPIRouter(name='tasks')
api.add_query_route(path='/open', query=open_work, method='get')
api.add_query_route(path='/inbox', query=inbox, method='get')
api.add_query_route(path='/approvals', query=approval_queue, method='get')

app = fastapi.FastAPI(title='agent-tasks')
app.include_router(api)

class EnqueueBody(BaseModel):
 title: str
 brief: str

class ClaimBody(BaseModel):
 agent_id: str
 task_id: str | None = None

class TaskBody(BaseModel):
 task_id: str
 agent_id: str

class DraftBody(TaskBody):
 draft: str

class NoteBody(BaseModel):
 task_id: str
 note: str | None = None

def _or_409(row: dict[str, Any] | None, detail: str) -> dict[str, Any]:
 if row is None:
 raise fastapi.HTTPException(status_code=409, detail=detail)
 return {
 'id': str(row['id']),
 'state': row['state'],
 'owner': row['owner'],
 'passed': row.get('passed'),
 'evidence': row['evidence'],
 }

@app.post('/enqueue')
def enqueue_route(body: EnqueueBody) -> dict[str, Any]:
 return _or_409(enqueue(body.title, body.brief), 'not enqueued')

@app.post('/claim')
def claim_route(body: ClaimBody) -> dict[str, Any]:
 return _or_409(claim_task(body.agent_id, body.task_id), 'no queued task, or it was already claimed')

@app.post('/start-draft')
def start_draft_route(body: TaskBody) -> dict[str, Any]:
 return _or_409(start_draft(body.task_id, body.agent_id), 'not your task, or not claimed or drafting')

@app.post('/request-approval')
def request_approval_route(body: DraftBody) -> dict[str, Any]:
 row = request_approval(body.task_id, body.agent_id, body.draft)
 if row is not None and row['state'] != AWAITING_APPROVAL:
 raise fastapi.HTTPException(status_code=409, detail=f"evidence failed: {row['evidence']}")
 return _or_409(row, 'not your task, or not drafting')

@app.post('/approve')
def approve_route(body: NoteBody) -> dict[str, Any]:
 return _or_409(approve(body.task_id, body.note), 'not awaiting approval')

@app.post('/reject')
def reject_route(body: NoteBody) -> dict[str, Any]:
 return _or_409(reject(body.task_id, body.note or ''), 'not awaiting approval')

@app.post('/complete')
def complete_route(body: TaskBody) -> dict[str, Any]:
 return _or_409(complete_task(body.task_id, body.agent_id), 'not approved, or not your task')

@app.post('/fail')
def fail_route(body: NoteBody) -> dict[str, Any]:
 return _or_409(fail_task(body.task_id, body.note or ''), 'already done or failed')
```

 
`request_approval` returns `None` for another agent's task or a task that is not drafting, and the written row, still in `drafting`, when the check fails. The route turns the second into a 409 that carries `evidence`. After a full HTTP pass, the OpenAPI paths were `/approvals`, `/approve`, `/claim`, `/complete`, `/enqueue`, `/fail`, `/inbox`, `/open`, `/reject`, `/request-approval`, `/start-draft`. The how-to for typed clients is [How to serve a typed AI endpoint](/blog/serve-typed-ai-endpoint-pixeltable) and the glossary page [What is HTTP serving?](/learn/what-is-http-serving).

 
## Same functions as MCP tools

 
Pixeltable 0.7.23 *consumes* MCP servers (`pxt.mcp_udfs`) so a computed column can call someone else's tools. It does not publish yours. Rosen describes an MCP server with tools such as `claim_task` and `complete_task`. That server is a thin wrapper around the Python you already have. Install it with `pip install mcp` (2.x, which renamed FastMCP to `MCPServer`) and save it as `mcp_server.py` next to `app.py`. It runs in its own process, so it binds the models itself. I called all five tools in-process with `MCPServer.call_tool`. A short draft left the row in `drafting`. A good draft moved it to `awaiting_approval`, and approve and complete left it `done` with owner `mcp-agent`. I did not run `mcp.run(transport='stdio')` or connect Claude Code or Codex to it.

 
```python
from mcp.server.mcpserver import MCPServer

from app import TableModel, approve, claim_task, complete_task, request_approval, start_draft

TableModel.bind_all('agent_sm')
mcp = MCPServer('agent-tasks')

def _summary(row: dict | None) -> dict | None:
 return None if row is None else {'id': str(row['id']), 'state': row['state'], 'owner': row['owner']}

@mcp.tool()
def claim_task_tool(agent_id: str, task_id: str | None = None) -> dict | None:
 return _summary(claim_task(agent_id, task_id))

@mcp.tool()
def start_draft_tool(task_id: str, agent_id: str) -> dict | None:
 return _summary(start_draft(task_id, agent_id))

@mcp.tool()
def request_approval_tool(task_id: str, agent_id: str, draft: str) -> dict | None:
 return _summary(request_approval(task_id, agent_id, draft))

@mcp.tool()
def approve_tool(task_id: str, note: str | None = None) -> dict | None:
 return _summary(approve(task_id, note))

@mcp.tool()
def complete_task_tool(task_id: str, agent_id: str) -> dict | None:
 return _summary(complete_task(task_id, agent_id))

if __name__ == '__main__':
 mcp.run(transport='stdio')
```

 
Point Claude Code or Codex at that stdio server, or keep calling the HTTP service. Either way the table is the shared machine. Agents that call other agents as tools, rather than sharing this queue, are a different pattern: [Pixelagent team workflows](/blog/pixelagent-team-workflows). [What is MCP?](/learn/what-is-mcp) is the glossary page. Pixeltable Cloud's hosted MCP at `/mcp/cloud` is a different product. It operates a signed-in organization. It is not this task queue.

 
## When to use Temporal, Restate, DBOS, or LangGraph

 
Rosen's tradeoff is the right one. A homegrown machine is a handful of states, some transition functions, and a polling loop. You keep the coding agent you already run. You keep the database you already have. You can also end up rebuilding checkpointing, locking, scheduling, observability, and crash recovery of the *loop*.

 

 - [Temporal](https://temporal.io/), [Restate](https://www.restate.dev/), and [DBOS](https://dbos.dev/) are durable-execution engines. Use them when a tool call must resume after the process dies, when you need timers and child workflows, or when the unit of recovery is a step in the agent loop.

 - [LangGraph](https://www.langchain.com/langgraph) is a checkpointed graph of LLM nodes. Use it when the workflow *is* that graph.

 - n8n, Mastra, and similar workflow runners are fine when the job is SaaS glue. The comparison against those for media pipelines is [The Pipeline Is the Table](/blog/ai-automation-workflow). Airflow is the job-DAG cousin: [Pixeltable vs Airflow](/blog/pixeltable-vs-airflow-ml-orchestration).

 

 
Pixeltable is the state and the data. Media, embeddings, and evals sit on the same rows as `state`. That is the [storage / orchestration / retrieval](/blog/triforce-storage-orchestration-retrieval) split, and the same "compute close to data" argument as [pg_durable](/blog/pg-durable-compute-close-to-data-multimodal-pixeltable), one layer up from SQL. It will not retry a hung HTTP call to a tool, and it will not schedule a timeout. Pair it with a durable-execution engine when you need both.

 
## The same file on Pixeltable Cloud

 
Once `pxt db status pxt://org:main` shows AVAILABLE, run `pxt db diff pxt://org:main --add-config-entry`, then `pxt db update pxt://org:main -f`, `pxt schema update app.py pxt://org:main -f`, and `pxt service update app.py pxt://org:main -f`. `pxt db update` errors until the `[[pixeltable.database]]` entry exists. I ran none of the hosted steps.The walkthrough is [Deploy to Pixeltable Cloud](https://docs.pixeltable.com/cloud). `pxt new` makes a free hosted database with no account. Configuration of that database is in [Configure Pixeltable Cloud](/blog/configure-pixeltable-cloud). The local-to-cloud loop for one `app.py` is [What is Pixeltable?](/blog/what-is-pixeltable-one-app-py-local-to-cloud).

 
Next: put the Tasks model, the transition functions and the HTTP routes in `app.py`, run the commands at the top, and call `POST /claim`. Run the history snippet in a REPL after `pxt schema update`, and keep the judge variant and `mcp_server.py` out of `app.py`. Or hand a coding agent [get-started.md](/get-started.md) and this post.
## Frequently asked questions

### What is an agent state machine?

An agent state machine is software that owns operational rules an LLM would otherwise keep in a prompt: which action is allowed now, when a human must approve, and when the work is done. The model proposes a transition. Code accepts or refuses it. The current state lives in a table, not in the conversation.

### How do you add human-in-the-loop approval to an AI agent workflow?

Store each unit of work as a row. The agent writes a draft and asks to move the row to awaiting_approval. A computed column scores that draft. Software moves the row only when the score passes. A reviewer calls approve or reject. Reject sets the row back to drafting with a note. The agent never writes the state column on its own.

### Does Pixeltable have compare-and-set or row locking for a task queue?

Pixeltable has no user-facing compare-and-set API. The guard is an update whose where clause names the id and the expected state, then a check of UpdateStatus.row_count_stats.upd_rows. Default tables are data-versioned, so a data write takes an exclusive table lock and writers serialize. A lost claim returns no row. Nested JSON is not SQL-filterable; store a bool column for the gate. A UUID filter needs a uuid.UUID, not a string.

### When should I use Temporal, Restate, DBOS, or LangGraph instead of Pixeltable?

Use Temporal, Restate, or DBOS when the problem is durable execution of the agent loop: retries, timers, child workflows, and crash recovery of an in-flight tool call. Use LangGraph when the work is a checkpointed graph of LLM nodes. Pixeltable is the state and data layer for the work itself. It is not a durable-execution engine. Keep Pixeltable next to those tools when the agents also write media, embeddings, or evals.

### How do Claude Code and Codex share one task queue?

Put the transitions in Python on a Pixeltable table, then expose claim_task, request_approval, and complete_task over HTTP with pxt service update, or wrap the same functions with an MCP server. Each coding agent calls the tools. Neither owns the state file.