API Reference

Yagra’s Python API typically builds an instance with Yagra.from_workflow(...) and executes it with invoke(...). The CLI entry point is main() (the yagra command).

Public API for the Yagra package.

class yagra.Yagra(compiled_graph, trace_collector=None)

Bases: object

Wrapper for executing LangGraph built from workflow YAML.

Parameters:

  • compiled_graph (CompiledStateGraph)

  • trace_collector (Any | None)

property compiled_graph: CompiledStateGraph

Returns the held compiled graph.

classmethod from_workflow(workflow_path, registry, bundle_root=None, state_schema=None, checkpointer=None, observability=False)

Creates a Yagra instance from a workflow file.

If interrupt_before / interrupt_after are defined in the workflow YAML, a checkpointer must be specified to enable HITL (Human-in-the-Loop) functionality. If checkpointer is None, the interrupt configuration is ignored.

Parameters:

  • workflow_path (str | PathLike[str]) – Path to the entry workflow.yaml.

  • registry (NodeRegistryPort | Mapping[str, NodeHandler]) – Registry implementation resolving handler names to callables, or handler mapping.

  • bundle_root (str | PathLike[str] | None) – Base directory for resolving split references. Defaults to workflow parent directory if not specified.

  • state_schema (Any) – LangGraph state schema. If None, automatically resolved from the workflow’s state_schema definition or defaults to dict.

  • checkpointer (BaseCheckpointSaver | None) – LangGraph checkpointer. Required when using interrupt_before / interrupt_after.

  • observability (bool) – If True, enables execution tracing (G-14, G-15). Each invoke() call records node-level traces in memory and can be read via get_last_trace(). Trace files are written to .yagra/traces/ only when trace=True is passed to invoke().

Returns:

Yagra instance containing the compiled graph.

Return type:

Yagra

get_last_trace()

Returns the trace captured during the most recent invoke() call.

Returns:

The last WorkflowRunTrace when observability is enabled and invoke() has been executed; otherwise None.

Return type:

WorkflowRunTrace | None

invoke(state, thread_id=None, trace=False, trace_dir=None)

Executes the graph with an initial state.

When using HITL (Human-in-the-Loop), specify thread_id. If interrupted by interrupt_before / interrupt_after, the return value is the state just before interruption. Call resume() to resume.

Parameters:

  • state (Mapping[str, Any]) – State dictionary at execution start.

  • thread_id (str | None) – Thread ID used for checkpoint management. Specify a unique string when using HITL. If None, state is not saved even if a checkpointer is configured.

  • trace (bool) – If True and observability=True was set in from_workflow(), writes a structured JSON trace to .yagra/traces/ after execution. In-memory trace capture is always enabled for observability=True, regardless of this flag.

  • trace_dir (str | Path | None) – Override the trace output directory. Defaults to .yagra/traces/ relative to cwd if not specified.

Returns:

State dictionary after execution (or at interruption point).

Raises:

TypeError – If the graph execution result is not dict-compatible.

Return type:

dict[str, Any]

resume(update=None, thread_id=None)

Resumes a graph that was interrupted.

Resumes a workflow that was interrupted by invoke(). You can modify the state with update before resuming (e.g., injecting human review results).

Parameters:

  • update (Mapping[str, Any] | None) – Update dict to merge into state before resuming. If None, resumes without modifying state.

  • thread_id (str | None) – Thread ID specified at invoke() time.

Returns:

State dictionary after resuming execution.

Raises:

  • TypeError – If the graph execution result is not dict-compatible.

  • ValueError – If thread_id is not provided when checkpointer requires it.

Return type:

dict[str, Any]

yagra.main()

CLI entry point for Yagra.

Return type:

None