Skip to content

Python API

Every name athanore exports, which is what a workflow module imports. Generated from athanore.__all__, with the file each name is defined in, its signature, and the first line of its docstring.

  • ACPAgent: class, athanore/agents/acp.py
  • ACPAgent(command: Sequence[str] | None = None, cwd: str | None = None, timeout: float | None = None, env: Mapping[str, str] | None = None, session_id: str | None = None)
  • An agent façade backed by an ACP subprocess.
  • Agent: class, athanore/agents/base.py
  • Agent()
  • The base façade: configuration, a prompt, and a run() to override.
  • AgentError: class, athanore/agents/base.py
  • An agent run, or its submission, is unusable.
  • AgentResult: class, athanore/agents/base.py
  • AgentResult(status: Literal['complete', 'failed'] = 'complete', output: Any = None, text: str = '', session_id: str | None = None, stop_reason: str | None = None, error: str | None = None, stats: dict[str, Any] = <factory>)
  • What one run() produced.
  • AgentSession: class, athanore/agents/acp.py
  • AgentSession(session_id: str, turn: Callable[[str], Awaitable[AgentResult]])
  • A session held open by :meth:ACPAgent.open: its id, and prompt().
  • GraphError: class, athanore/graph/builder.py
  • Raised for invalid workflow graphs, at finalization or routing time.
  • NonRetryable: class, athanore/engine/errors.py
  • Raised by a node body that knows a retry cannot help.
  • PluginContext: class, athanore/plugins/context.py
  • PluginContext(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, _ops: Ops | None = None)
  • The one argument a plugin handler takes.
  • PluginError: class, athanore/plugins/decl.py
  • PluginError(status: int, message: str)
  • What a plugin handler raises to answer with a status of its own.
  • Pool: class, athanore/engine/pools.py
  • Pool(name: str, capacity: int)
  • A named concurrency cap, as a user writes it.
  • Server: class, athanore/server.py
  • Server(settings: AthanoreSettings | None = None)
  • One process serving the workflows registered on it.
  • Workflow: class, athanore/workflow.py
  • Workflow(name: str, *, assets: str | Path | None = None)
  • One workflow: a name, its nodes, and the graph they finalize into.
  • __version__: str, athanore/__init__.py
  • current_task: function, athanore/engine/context.py
  • current_task() -> TaskContext
  • The bound :class:TaskContext, or raise.
  • human_input: function, athanore/requests/human.py
  • human_input(prompt: str, *, options: Sequence[Option] | None = None, output_model: type[BaseModel] | None = None, timeout: float | None = None) -> Any
  • Ask the operator prompt and return their answer.
  • maybe_current_task: function, athanore/engine/context.py
  • maybe_current_task() -> TaskContext | None
  • The bound :class:TaskContext, or None outside an attempt.