Workflow
Workflows, built from a description and run from code.
brief = noclick.Workflow.create("Morning brief")
brief.build("Every day at 8, email me today's calendar").wait()
run = brief.run().wait()Your keys read every workflow your account can open in the app, its own and
those shared with it (access), and change, build and run the ones it owns
and pays for. NoClick's builder (the one in the app) builds and changes one
from what you describe: a WorkflowBuild you wait on, which may stop to ask
something first, and which can publish the workflow's app too. A run starts
where its trigger does, with the inputs as the trigger's payload. A run started
inside as_user (or with user=) is recorded for that user and counts
toward their spend cap; its steps still use the app's connections.
class WorkflowA workflow built in the NoClick app: its trigger, steps and outputs. Run it
with run(inputs); inputs names what its trigger takes. access is
your account's: owner, editor or viewer (one shared with it);
changing, building and running take owning it and paying for it.
- Like every live object, it also has hydrate(), refresh() and raw.
| id | str | |
| name | str | |
| description | str | None | |
| inputs | dict[str, Any] | A JSON schema of what |
| app_url | str | None | Where to open it in the NoClick app. |
| access | str | Your account's access: |
| created_at | Any | |
| updated_at | Any | |
| steps | list[WorkflowStep] | Its steps in order, each with its settings, connection and trigger state (read with one workflow, not in a list). |
| edges | list[dict[str, Any]] | How its steps are wired: |
| paused | bool | Whether a pause switched its triggers off. |
| readiness | dict[str, Any] | Whether it's set up to run: |
| urls | list[dict[str, Any]] | The URLs it answers at: |
| latest_build | WorkflowBuild | None | The build |
createclassmethodasync
noclick.Workflow.create(
name, *, instructions=None, description=None, client=None
) -> WorkflowA new workflow in your account. With instructions NoClick's
builder starts building it: wait on latest_build. A name your
account already has raises noclick.Conflict (change that one with
build).
NoneNoneThe client to send with; defaults to the module client built from NOCLICK_API_KEY.
None)The new workflow; latest_build is its build when instructions describe it.
brief = noclick.Workflow.create(
"Morning brief", instructions="Every day at 8, email me today's calendar and unread emails",
)
result = brief.latest_build.wait(timeout=900)
print(result.summary, [step.name for step in result.steps])from_nameclassmethod
noclick.Workflow.from_name(name, *, client=None) -> WorkflowA lazy reference to the workflow called name in the app. Two
workflows with the same name raise noclick.Conflict on first use
(rename one, or use from_id).
The client to send with; defaults to the module client built from NOCLICK_API_KEY.
None)A workflow that loads the first time it's used. Nothing is sent yet.
intake = noclick.Workflow.from_name("order-intake")
run = intake.run({"order_id": "o_1042"}).wait()
print(run.outputs)from_idclassmethod
noclick.Workflow.from_id(workflow_id, *, client=None) -> WorkflowA lazy reference to a workflow by id.
The client to send with; defaults to the module client built from NOCLICK_API_KEY.
None)A workflow that loads the first time it's used. Nothing is sent yet.
intake = noclick.Workflow.from_id("2e3a9296-1bf5-4088-a20d-7c31937380c4")
print(intake.name, intake.inputs)getclassmethodasync
noclick.Workflow.get(workflow_id, *, client=None) -> WorkflowLoad a workflow by id now; an unknown id raises noclick.NotFound.
The client to send with; defaults to the module client built from NOCLICK_API_KEY.
None)The workflow, loaded.
intake = noclick.Workflow.get("2e3a9296-1bf5-4088-a20d-7c31937380c4")
print(intake.name, intake.app_url)listclassmethodasync
noclick.Workflow.list(*, q=None, limit=None, client=None) -> list[Workflow]Your account's workflows, most recently changed first. q is a
case-insensitive prefix of the name.
NoneReturn at most this many.
None)The client to send with; defaults to the module client built from NOCLICK_API_KEY.
None)Every matching workflow, fetched page by page (or the first limit).
for workflow in noclick.Workflow.list():
print(workflow.name, workflow.updated_at)list_pageclassmethodasync
noclick.Workflow.list_page(*, cursor=None, limit=None, q=None, client=None) -> PageOne page of workflows; pass next_cursor back as cursor.
The next_cursor of the page before; omit it for the first page.
None)Return at most this many.
None)NoneThe client to send with; defaults to the module client built from NOCLICK_API_KEY.
None)A page whose data holds the workflows and whose next_cursor is None on the last page.
page = noclick.Workflow.list_page(limit=25)
for workflow in page.data:
print(workflow.name)buildasync
workflow.build(instructions=None, *, publish=None) -> WorkflowBuildChange the workflow by describing the change; NoClick's builder makes
it, reading the workflow as it stands. publish=True then puts its
app live (or pass {"subdomain": "morrow", "title": ...}; publish()
and unpublish() publish without a change). One build at a time:
another in progress raises noclick.Conflict. A change costs AI
usage, as building in the app does.
NoneNoneA WorkflowBuild: wait() gives the WorkflowBuildResult (its steps, and what was
published), and next_question() returns anything it asks first.
brief = noclick.Workflow.from_name("Morning brief")
result = brief.build("Also post the brief to #team on Slack").wait(timeout=900)
print(result.summary)publishasync
workflow.publish(*, subdomain=None, title=None) -> WorkflowBuildPut the workflow's app live as it now stands, at subdomain
(morrow is morrow.noclick.app; omitted, the address it has).
NoneNoneA WorkflowBuild whose result's published holds the app's url.
site = noclick.Workflow.from_name("Morrow Bakehouse site")
print(site.publish(subdomain="morrow").wait().published["url"])unpublishasync
workflow.unpublish() -> WorkflowBuildTake the workflow's app down: its public URL stops answering.
A WorkflowBuild; its result's published says unpublished.
noclick.Workflow.from_name("Morrow Bakehouse site").unpublish().wait()pauseasync
workflow.pause(reason=None) -> WorkflowSwitch off each of its triggers that's on, with reason beside them
in the app: nothing starts it on its own until you resume(). Runs
from code still start.
NoneThis workflow, read again (paused is True).
digest = noclick.Workflow.from_name("Hourly digest")
digest.pause("Too noisy over the weekend")resumeasync
workflow.resume() -> WorkflowSwitch back on each trigger a pause switched off (never one turned off by hand).
This workflow, read again.
noclick.Workflow.from_name("Hourly digest").resume()restoreasync
workflow.restore() -> WorkflowBring the workflow back from the trash (within 30 days of delete());
its schedules and webhooks register again.
This workflow, read again.
noclick.Workflow.from_id("2e3a9296-1bf5-4088-a20d-7c31937380c4").restore()deleteasync
workflow.delete() -> NoneMove the workflow to the trash: its schedules and webhooks stop now,
and restore() brings it back within 30 days.
Nothing.
noclick.Workflow.from_name("Old digest").delete()runasync
workflow.run(inputs=None, *, user=..., idempotency_key=None) -> OperationStart a run with inputs (checked against inputs, the workflow's
schema; a mismatch raises noclick.InvalidRequest naming the field).
What would stop the run is refused before anything runs: a step with no
account connected raises noclick.NeedsConnection, and a workflow that
can't start from code (several triggers and no Run trigger, or only
events an integration pushes) or isn't set up noclick.ActionRequired.
NoneWho the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).
Makes a retried create safe: the same key returns the first result. Generated when omitted.
None)An operation whose .wait() gives the finished WorkflowRun, or
raises the run's error if it failed.
Run it and wait for the outputs:
intake = noclick.Workflow.from_name("order-intake")
run = intake.run({"order_id": "o_1042", "rush": True}).wait(timeout=300)
print(run.outputs["summary"])Record the run for a user, and make a retried request safe:
intake = noclick.Workflow.from_name("order-intake")
op = intake.run({"order_id": "o_1042"}, user="acme", idempotency_key="order-o_1042")
print(op.id) # resume anywhere with noclick.Operation.from_id(op.id).wait()runsasync
workflow.runs(*, status=None, limit=20, user_filter=None, user=...) -> list[WorkflowRun]Recent runs, newest first, however they started (your code, its
endpoint, a schedule, the app). Outside as_user you read runs recorded
for your account; user_filter reads your users'.
NoneReturn at most this many.
20)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.
None)Who the call acts for. Defaults to the user set by noclick.as_user (none: the account itself).
The first limit matching runs (20 by default; None for all of them).
intake = noclick.Workflow.from_name("order-intake")
for run in intake.runs(status="failed", user_filter=noclick.ALL_USERS):
print(run.started_at, run.source, run.error)get_runasync
workflow.get_run(run_id) -> WorkflowRunOne run, with what each step did (steps: status, error and output, in order).
The WorkflowRun; an unknown id raises noclick.NotFound.
intake = noclick.Workflow.from_name("order-intake")
failed = intake.runs(status="failed", limit=1)[0]
for step in intake.get_run(failed.id).steps:
print(step.name, step.status, step.error)endpointasync
workflow.endpoint(*, public=None) -> WorkflowEndpointThe workflow's HTTP endpoint, turned on if it wasn't. public=True
lets a publishable key call it from a web page, with per-visitor rate
limits; public=False requires a secret key again; leave it out to keep
what it is.
NoneThe endpoint: its url, whether it's public, and the method to call it with.
intake = noclick.Workflow.from_name("order-intake")
endpoint = intake.endpoint(public=True)
print(endpoint.method, endpoint.url, endpoint.public)WorkflowRun
class WorkflowRunOne run of a workflow. status is running (a run waiting on an
approval or a delay too), succeeded or failed (error says why).
outputs holds what its final steps produced (each step that finished
with nothing after it running), by step name. source is how it started:
api, endpoint, schedule, trigger (an event, like a webhook
or an email) or app; inputs and operation_id are set for a run
started from code. steps says what each step did, read with one run
(Workflow.get_run); empty in a list.
| id | str | |
| workflow_id | str | None | |
| status | str | |
| inputs | dict[str, Any] | |
| outputs | dict[str, Any] | |
| error | str | None | |
| source | str | None | |
| user_id | str | None | |
| operation_id | str | None | |
| started_at | Any | |
| finished_at | Any | |
| steps | list[RunStep] | |
| raw | dict[str, Any] | The response JSON exactly as the API sent it. |
RunStep
class RunStepWhat one step did in a run: status is succeeded, failed
(error says why) or skipped; output is what it produced, with
long lists and text cut short.
| id | str | |
| name | str | |
| status | str | |
| error | str | None | |
| output | Any |
WorkflowEndpoint
class WorkflowEndpointThe URL a workflow answers at: send method (POST) with the inputs
as a JSON body, and the run's outputs come back (a run longer than 30
seconds answers 202 with its operation). A public endpoint also takes
a publishable key (nk_pub_...) that lists it, so a web page can call it;
otherwise it needs a secret key.
| url | str | |
| public | bool | |
| method | str | |
| raw | dict[str, Any] | The response JSON exactly as the API sent it. |
WorkflowBuild
class WorkflowBuild(Operation)A build of a workflow from a description (kind workflow.build). It's
an Operation: wait() returns the WorkflowBuildResult, or raises
why the build failed.
While it builds, step is building. When the builder needs something
it's awaiting_human (step asking) and question says what:
answer it with answer(), or send a person to its answer_url.
wait() waits through a question until someone answers it;
next_question() returns as soon as there is one.
- Inherits from Operation.
| workflow_id | str | None | |
| question | BuildQuestion | None | What the build waits on ( |
next_questionasync
workflow_build.next_question(timeout=None) -> BuildQuestion | NoneBlock until the build asks something (its question) or finishes (None: wait() gives the result).
NoneThe question to answer, or None once the build is done.
build = noclick.Workflow.from_name("Morning brief").build("Also post it to #team on Slack")
while (question := build.next_question(timeout=600)) is not None:
if not question.answerable_from_code:
print("A person answers this:", question.answer_url)
break
build.answer({question.inputs[0]["id"]: "#team"})
print(build.wait().summary)answerasync
workflow_build.answer(values=None, *, message=None, skip=False) -> WorkflowBuildAnswer the question the build waits on, by input id; message tells
the builder anything else, skip=True goes on without answering. An
input only a person can answer (an account to connect, secret values)
raises noclick.InvalidRequest naming answer_url.
NoneNoneFalseThis build, running again.
build = noclick.Workflow.from_name("Morning brief").build("Send it on Slack too")
question = build.next_question()
if question and question.answerable_from_code:
build.answer({"channel": "#team"}, message="Keep it under five lines")
print(build.wait().steps)WorkflowBuildResult
class WorkflowBuildResultA finished build: what the builder says it did (summary), the
workflow's steps as they now stand, and, when it published the
workflow's app, published ({state, url}: state is
published, renamed or unpublished).
| workflow_id | str | None | |
| summary | str | None | |
| steps | list[WorkflowStep] | |
| app_url | str | None | |
| published | dict[str, Any] | None | |
| raw | dict[str, Any] | The response JSON exactly as the API sent it. |
WorkflowStep
class WorkflowStepOne step of a workflow, as the app shows it: its name, its kind
(type, like automation-gmail or trigger-cron) and the
operation it does there (None for a step with one thing to do).
Read with one workflow, a step also says whether it's enabled, its
settings (never a secret; long values cut short), its connection
(None for a step that takes no account, else {attached, healthy, hint}: healthy is None when NoClick can't tell) and, for a
trigger, its trigger state ({error, paused_reason, next_run};
None for any other step).
| id | str | |
| name | str | |
| type | str | |
| operation | str | None | |
| enabled | bool | |
| settings | dict[str, Any] | |
| connection | dict[str, Any] | None | |
| trigger | dict[str, Any] | None |
BuildQuestion
class BuildQuestionWhat a build waits on. Each of inputs is {id, label, type, description, options?}; answer them with build.answer({id: value})
where answerable_from_code. An input of type credential (an account
to connect) or env (secret values) needs a person: send them to
answer_url, and the build goes on once they answer.
| id | str | None | |
| title | str | None | |
| inputs | list[dict[str, Any]] | |
| answer_url | str | None | |
| answerable_from_code | bool | |
| raw | dict[str, Any] | The response JSON exactly as the API sent it. |