Skip to content

Plugins

What a workflow declares to reach the interface, and what a handler is handed when one of those declarations runs. Generated from the declaration dataclasses and from PluginContext. How to write one is in Writing a plugin.

Declarations

Route

  • path: str
  • methods: tuple[str, ...]
  • fn: Callable[..., Any]

Action

  • name: str
  • title: str
  • scope: Scope
  • confirm: bool
  • model: type[BaseModel] | None
  • fn: Callable[..., Any]

Panel

  • name: str
  • slot: Slot
  • placement: Placement
  • kind: PanelKind
  • scope: Scope
  • node: str | None = None
  • source: Callable[..., Any] | str | None = None
  • element: str | None = None
  • refresh_on: tuple[str, ...] = ()

Handler

  • event: str
  • fn: Callable[..., Any]

Slot (and Scope, which is the same enum)

Where a panel is shown, and what a declaration needs resolved before it runs.

  • run
  • task
  • node
  • workflow
  • global_ = global

PanelKind

The renderer vocabulary: which of these a panel declares fixes the shape its source has to return.

  • markdown
  • kv
  • table
  • log
  • chart
  • dashboard
  • form
  • custom

Placement

  • card
  • pane

Methods a route may declare

  • DELETE
  • GET
  • HEAD
  • OPTIONS
  • PATCH
  • POST
  • PUT

PluginError

  • PluginError(status: int, message: str)
  • What a plugin handler raises to answer with a status of its own.

PluginContext

The one argument every plugin handler takes. Which of these a handler may reach for depends on the scope it was resolved in: a run-scoped declaration has a run, a task-scoped one has a task, and a workflow-scoped one has neither.

Attributes

  • workflow: str
  • services: PluginServices
  • run_id: str | None = None
  • task_id: int | None = None
  • node: str | None = None
  • run: RunRow | None = None
  • task: TaskRow | None = None

ctx.ops

  • The operator operations of 04, unscoped by design.

Ops (athanore/engine/ops.py)

  • await append_log(run_id: str, text: str) -> LogEntryRow
  • Add an operator note to a run's work log.
  • await cancel(run_id: str) -> list[int]
  • End a run and everything outstanding under it.
  • await delete(run_id: str) -> None
  • Cancel a run, then remove it and every row under it.
  • await edit(run_id: str, title: str | None = None, description: str | None = None) -> RunRow
  • Change a run's title, its description, or both.
  • await move(task_id: int, node: str) -> TaskRow
  • Cancel a task and enqueue its work at another node.
  • await pause(run_id: str) -> RunRow
  • Stop a run dispatching; let what is already running finish.
  • await reorder(run_id: str, direction: int | None = None, index: int | None = None) -> int
  • Move a run in the dispatch list, and return its position.
  • await rerun(run_id: str, node: str) -> TaskRow
  • Run a node again, with the payload and branch it last had.
  • await resume(run_id: str) -> RunRow
  • Let a paused run dispatch again.
  • await retry(task_id: int) -> TaskRow
  • Queue another attempt of a task that has stopped.
  • await set_status(task_id: int, status: SettableStatus) -> TaskRow
  • Put a task in one of the three statuses an operator may write.
  • await submit(workflow: str, title: str, description: str = '') -> RunRow
  • Queue a run of workflow and its one start task.

ctx.services

Each is a property, so reaching for one out of scope raises where it is reached for.

  • ctx.services.log → LogService
  • The work log of the attempt in scope.
  • ctx.services.stream → StreamService
  • The transcript of the attempt in scope.
  • ctx.services.submissions → SubmissionService
  • What the attempt in scope submitted, and the verbs over it.
  • ctx.services.requests → RequestsPort
  • The questions the attempt in scope has open.
  • ctx.services.run → PluginRuns
  • The run in scope, and this workflow's runs.
  • ctx.services.events → EventPort
  • Publishing plugin.<workflow>.<name>, and nothing else.

LogService (athanore/engine/services.py)

  • await append(text: str, *, author: LogAuthor | str = LogAuthor.agent, kind: LogKind | str | None = None) -> LogEntryRow
  • Add one entry to the run's work log and announce it.

StreamService (athanore/engine/services.py)

  • await append(kind: ChunkKind | str, text: str) -> int
  • Buffer one chunk and return the seq it was given.
  • await close() -> None
  • Stop the flusher and write what is left. Idempotent.
  • await flush() -> None
  • Write what has accumulated, now, without ending the transcript.

SubmissionService (athanore/engine/services.py)

  • await accept(payload: Any) -> SubmissionRow
  • Record a payload that passed the declared model.
  • await latest() -> SubmissionRow | None
  • The newest submission of this attempt, or None.
  • await reject(errors: Sequence[Mapping[str, Any]], schema: Mapping[str, Any]) -> dict[str, Any]
  • Announce a submission that did not fit, storing no row.
  • await repair(turn: int, reason: Literal['nothing_submitted', 'rejected']) -> Event
  • Announce a repair turn: the façade is asking again.

RequestsPort (athanore/engine/services.py)

  • await answer_as_engine(request_id: int, option_id: str | None = None) -> AnswerRow
  • await create_agent_request(prompt: str, *, mode: RequestMode | str, kind: RequestKind | str, options: list[dict[str, Any]] | None = None, schema: dict[str, Any] | None = None, tool_call: dict[str, Any] | None = None) -> RequestRow
  • Open a request the agent raised mid-turn.
  • await poll(request_id: int, wait_s: float) -> AnswerRow | None
  • register_schema_validator(request_id: int, schema: dict[str, Any]) -> None
  • Validate this request's form answer against schema.
  • register_validator(request_id: int, fn: AnswerValidator) -> None
  • Validate this request's form answer with fn when it lands.
  • await reopen_or_create(prompt: str, *, mode: RequestMode, kind: RequestKind, options: list[dict[str, Any]] | None = None, schema: dict[str, Any] | None = None) -> RequestRow
  • unregister_validator(request_id: int) -> None
  • Forget this request's validator. Not having one is not an error.
  • await wait(request_id: int, timeout: float | None = None) -> AnswerRow

PluginRuns (athanore/plugins/context.py)

  • await detail() -> RunDetailRows
  • The run in scope, its attempts, and their summed stats.
  • await events(after: int = 0, limit: int | None = None) -> list[EventRow]
  • The stored events of the run in scope, oldest first.
  • await get() -> RunRow
  • The run this context is scoped to, freshly read.
  • await list(status: str | None = None) -> list[RunSummary]
  • This workflow's runs, in list order.
  • await log_entries(after: int = 0, limit: int | None = None) -> list[LogEntryRow]
  • The work log of the run in scope, oldest first.

EventPort (athanore/engine/services.py)

  • await publish(name: str, data: Mapping[str, Any]) -> Event
  • Store and publish plugin.<workflow>.<name>.

Refusals

What the context itself raises when a handler reaches for something its scope does not have. Both are PluginError(400, ...), and an out-of-scope id is a 404 rather than a 403.

  • no run in scope
  • no task in scope