Operation

Operations: work that outlives a request (a human's answer, a connect link, an agent turn). op.wait() returns the typed outcome for its kind.

class Operation

A long-running piece of work. Block on it with wait().

Like every live object, it also has hydrate(), refresh() and raw.

Attributes

id	str
kind	str
status	str
step	str \| None	Where a running operation is. A turn of an agent on a CLI harness: `delivering` (starting its sandbox and handing it the message) or `working` (the harness is on it). A number's port (`phone.port_in`, invite-only beta): `submitted` (with the current carrier) or `scheduled` (it moves at `port_at`). `None` otherwise.
result	dict[str, Any] \| None	The raw result once succeeded.
created_at	Any
updated_at	Any
summary_text	str \| None	A short line saying what it is (a message preview, an addressee, a command).
user_id	str \| None
key_id	str \| None	The key that started it.
agent_id	str \| None
input	dict[str, Any] \| None	What it was started with (single-operation views, account principals).
started_at	Any
finished_at	Any
wait_ms	int \| None	Time queued before it ran.
run_ms	int \| None	Time running.
held_ms	int \| None	Time waiting on a person.
progress	dict[str, Any] \| None	What the work reported as it went, or `None` before it reported anything: its `step`, its `stages` (each `{name, status, started_at, ended_at}`, `status` being `done`, `current` or `pending`; a deploy's are Queued, Building, Shipping and Live), the last 50 `lines` it printed (a build's log, a call's turns, an agent's tool calls), `line_count` (every line so far) and `updated_at`. An agent that started it hears each report as an `operation` event.
undoable	bool \| None	Whether `undo()` can take it back now: a reversible write (a deletion that went to the trash, a dict's clear, a database migration) not undone yet. `None` for kinds that record no undo.
undone_at	Any	When it was undone, if it was.
transitions	list[OperationTransition]	Every status change, oldest first (empty for operations recorded before they were kept).
console_url	str \| None	This operation in the developer console (None off the hosted API).
error	Error \| None	The failure as an `noclick.Error` (None unless it failed).
done	bool	Whether the operation reached a terminal status.

listclassmethodasync

noclick.Operation.list(
    *, kind=None, status=None, status_not=None, agent=None, key=None,
    created_after=None, created_before=None, updated_after=None, limit=None,
    user_filter=None, user=..., client=None
) -> list[Operation]

Operations matching the filters, newest first (or the first limit). status="live" means every non-terminal status; key is a key id or prefix (nk_live_ab12); agent an Agent or id.

Parameters

kindManydefault None

statusManydefault None

status_notManydefault None

agentAnydefault None

keystr | Nonedefault None

created_afterAnydefault None

created_beforeAnydefault None

updated_afterAnydefault None

limitint | None

Return at most this many.

(default None)

user_filterAny

Which users a read covers: noclick.ALL_USERS, noclick.NO_USER, a User, or your id for one. It filters without acting as that user.

(default None)

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

Returns

list[Operation]

Every matching operation, fetched page by page.

Usage

See what failed in the last day:

from datetime import datetime, timedelta, timezone

yesterday = datetime.now(timezone.utc) - timedelta(days=1)
for op in noclick.Operation.list(status="failed", created_after=yesterday):
    print(op.kind, op.error)

Or the turns one agent is still working on:

support = noclick.Agent.from_name("support")
live = noclick.Operation.list(kind="agent.message", status="live", agent=support)

A read acts for the current user; user_filter reads across them:

failed = noclick.Operation.list(status="failed", user_filter=noclick.ALL_USERS, limit=20)

list_pageclassmethodasync

noclick.Operation.list_page(
    *, cursor=None, limit=None, kind=None, status=None, status_not=None, agent=None,
    key=None, created_after=None, created_before=None, updated_after=None,
    user_filter=None, user=..., client=None
) -> Page

One page of matching operations; pass next_cursor back as cursor.

Parameters

cursorstr | None

The next_cursor of the page before; omit it for the first page.

(default None)

limitint | None

Return at most this many.

(default None)

kindManydefault None

statusManydefault None

status_notManydefault None

agentAnydefault None

keystr | Nonedefault None

created_afterAnydefault None

created_beforeAnydefault None

updated_afterAnydefault None

user_filterAny

Which users a read covers: noclick.ALL_USERS, noclick.NO_USER, a User, or your id for one. It filters without acting as that user.

(default None)

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

Returns

Page

A page whose data holds the operations and whose next_cursor is None on the last page.

Usage

page = noclick.Operation.list_page(status="failed", limit=100)
while True:
    for op in page.data:
        print(op.id, op.kind, op.error)
    if page.next_cursor is None:
        break
    page = noclick.Operation.list_page(status="failed", limit=100, cursor=page.next_cursor)

summaryclassmethodasync

noclick.Operation.summary(
    *, kind=None, status=None, status_not=None, agent=None, key=None,
    created_after=None, created_before=None, updated_after=None, user_filter=None,
    user=..., client=None
) -> OperationSummary

How many operations match the same filters, by status.

Parameters

kindManydefault None

statusManydefault None

status_notManydefault None

agentAnydefault None

keystr | Nonedefault None

created_afterAnydefault None

created_beforeAnydefault None

updated_afterAnydefault None

user_filterAny

Which users a read covers: noclick.ALL_USERS, noclick.NO_USER, a User, or your id for one. It filters without acting as that user.

(default None)

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

Returns

OperationSummary

The total, and in by_status a count for every status, zeros included.

Usage

counts = noclick.Operation.summary(kind="agent.message", status="live")
print(counts.total, counts.by_status["running"], counts.by_status["awaiting_human"])

from_idclassmethod

noclick.Operation.from_id(operation_id, *, client=None, user=...) -> Operation

A lazy reference to an operation by id.

Parameters

operation_idstrrequired

clientClient | None

The client to send with; defaults to the module client built from NOCLICK_API_KEY.

(default None)

userAny

Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).

(defaults to the current user)

Returns

Operation

An operation that loads the first time you read a field. Nothing is sent yet.

Usage

Operations outlive your process, so keep the id and pick it up anywhere:

op = noclick.Operation.from_id("3b9d6f2e-1c4a-4e8b-9f7d-2a6c5e8b1d40")
reply = op.wait()

Or check on it without waiting:

op = noclick.Operation.from_id("3b9d6f2e-1c4a-4e8b-9f7d-2a6c5e8b1d40").hydrate()
print(op.status, op.done)

wait async

operation.wait(timeout=None) -> Any

Block until the operation finishes.

Raises the operation's error if it failed, OperationCanceled / OperationExpired, or OperationTimeout after timeout seconds.

Parameters

timeoutfloat | Nonedefault None

Returns

Any

The outcome, typed for the operation's kind: an AgentReply from agent.send, True or False from human.approve, the connected Connection from a connect link. A kind the client has no type for gives the raw result.

Usage

Start an agent turn, then block on it:

agent = noclick.Agent.from_name("support")
turn = agent.send("Draft the monthly report", user="user_42")
reply = turn.wait(timeout=120)
print(reply.text)

A timeout leaves the operation running, so you can wait on it again later:

op = noclick.Operation.from_id("3b9d6f2e-1c4a-4e8b-9f7d-2a6c5e8b1d40")
try:
    reply = op.wait(timeout=30)
except noclick.OperationTimeout:
    print(op.status)   # still running

tool_calls async

operation.tool_calls() -> list[ToolCall]

The tool calls an agent.message turn made, oldest first (none for other kinds).

Returns

list[ToolCall]

Each call's tool, arguments, status, duration and a preview of its result.

Usage

agent = noclick.Agent.from_name("support")
turn = agent.send("Find last week's invoices", user="user_42")
turn.wait()
for call in turn.tool_calls():
    print(call.tool_name, call.status, call.duration_ms)

events iterator

operation.events(*, interval=1.0) -> Iterator[OperationEvent]

Yield an OperationEvent for each status or step change until it finishes.

Parameters

intervalfloatdefault 1.0

Returns

Iterator[OperationEvent]

The changes, polled every interval seconds: first the status (and step) it has now, last its final one. An event's result is the raw result, not the typed outcome wait() returns.

Usage

Print each change as it happens, a CLI agent's turn going from delivering to working:

op = noclick.Operation.from_id("3b9d6f2e-1c4a-4e8b-9f7d-2a6c5e8b1d40")
for event in op.events(interval=5):
    print(event.status, event.step, event.updated_at)

In async code, iterate the .aio() form:

op = noclick.Operation.from_id("3b9d6f2e-1c4a-4e8b-9f7d-2a6c5e8b1d40")
async for event in op.events.aio():
    print(event.status)

cancel async

operation.cancel() -> Operation

Ask the server to cancel.

Returns

Operation

This operation, now canceled.

Usage

Cancel a turn you no longer need:

agent = noclick.Agent.from_name("support")
turn = agent.send("Draft the monthly report", user="user_42")
turn.cancel()
print(turn.status)   # canceled

An operation that already finished can't be canceled:

op = noclick.Operation.from_id("3b9d6f2e-1c4a-4e8b-9f7d-2a6c5e8b1d40")
try:
    op.cancel()
except noclick.Conflict as err:
    print(err.operation_status)   # succeeded, failed, ...

undo async

operation.undo() -> dict[str, Any]

Take back a reversible write, once: a deletion comes out of the trash, a dict's cleared keys come back, a migrated database is restored to the moment before, a promote puts production back on the release before with the app's databases as they were. What reached the world (an email sent, a payment) has no undo: that raises noclick.Conflict, as does undoing twice.

Returns

dict[str, Any]

What the undo put back, as its own API shows it (a restore of a database is an operation of its own).

Usage

item = noclick.Dict.from_name("rates-cache").delete()
noclick.Operation.from_id(item.operation_id).undo()

OperationEvent

class OperationEvent

A status or step change observed by Operation.events(): step is where a running operation is (delivering or working for a CLI agent's turn, submitted or scheduled for a port; None otherwise).

Attributes

status	str
step	str \| None
updated_at	Any
result	Any
error	Error \| None
raw	dict[str, Any]	The response JSON exactly as the API sent it.

OperationSummary

class OperationSummary

How many operations match, by status.

Attributes

total	int
by_status	dict[str, int]
raw	dict[str, Any]	The response JSON exactly as the API sent it.

OperationTransition

class OperationTransition

One status change the server recorded (by names who canceled).

Attributes

status	str
at	Any
by	str \| None