PES API Reference

Three-phase (plan → execute → summarize) execution engine base class.

Subclasses customize behavior by overriding _call_sdk_query and four extension points: _build_plan_prompt, _build_execute_prompt, _build_summarize_prompt, and _post_summarize.

Parameters:

Name	Type	Description	Default
`config`	`PESConfig`	PES configuration loaded via `load_pes_config()`.	required
`model`	`ModelConfig`	LLM provider configuration.	required
`workspace`	`WorkspaceHandle`	Isolated workspace handle for this run.	required
`hooks`	`HookManager \| None`	Optional hook manager for lifecycle callbacks.	`None`
`trajectory_store`	`TrajectoryStore \| None`	Optional trajectory store for run recording.	`None`
`runtime_context`	`dict[str, Any] \| None`	Extra parameters passed to extension points (e.g., `output_schema` for ExtractorPES).	`None`
`llm_client`	`LLMClient \| None`	Optional pre-configured LLM client (created automatically if not provided).	`None`

Example

from scrivai import BasePES, ModelConfig, load_pes_config class MyPES(BasePES): ... pass # override extension points as needed pes = MyPES( ... config=load_pes_config(Path("my_pes.yaml")), ... model=ModelConfig(model="claude-sonnet-4-20250514"), ... workspace=ws, ... ) run = await pes.run("Process this document")

`run(task_prompt)` `async`

Execute the three-phase pipeline and return the completed run.

Runs plan → execute → summarize sequentially. Each phase validates its file-contract outputs and retries on failure (up to max_retries configured per phase).

Parameters:

Name	Type	Description	Default
`task_prompt`	`str`	Natural language description of the task.	required

Returns:

Type	Description
`PESRun`	A `PESRun` with `status='completed'` on success, or
`PESRun`	`status='failed'` with `error` details on failure.

Raises:

Type	Description
`PhaseError`	If a phase exhausts all retries.

`plan(run, task_prompt)` `async`

Entry point for the plan phase.

`execute_phase(run, task_prompt)` `async`

Entry point for the execute phase.

`summarize(run, task_prompt)` `async`

Entry point for the summarize phase.

`build_execution_context(phase, run)` `async`

Build the execution context for this phase. Returns empty dict by default.

`build_phase_prompt(phase, phase_cfg, context, task_prompt)` `async`

Render the prompt via PromptManager templates.

Override this method for fully custom prompt assembly. For most cases, override build_execution_context instead.

`postprocess_phase_result(phase, result, run)` `async`

Post-process the LLM response. Default is a no-op. Exceptions become response_parse_error (not retryable).

`validate_phase_outputs(phase, phase_cfg, result, run)` `async`

Validate required output files. Default validates each rule in required_outputs.

Exceptions become output_validation_error (retryable).

Bases: BasePES

Extract structured data from documents using LLM.

Inherits from BasePES with no additional constructor parameters. Business parameters are passed via runtime_context.

Parameters:

Name	Type	Description	Default
`config`	`PESConfig`	PES configuration (use `load_pes_config("extractor.yaml")`).	required
`model`	`ModelConfig`	LLM provider configuration.	required
`workspace`	`WorkspaceHandle`	Isolated workspace for this run.	required
`runtime_context`	`dict[str, Any] \| None`	Must include: - `output_schema` (`type[BaseModel]`): Pydantic model for output validation.	`None`

Phase contracts

plan → working/plan.json with extraction items
execute → working/findings/<id>.json per planned item
summarize → working/output.json conforming to output_schema

Example

from pydantic import BaseModel from scrivai import ExtractorPES, ModelConfig, load_pes_config class Items(BaseModel): ... items: list[str] pes = ExtractorPES( ... config=load_pes_config(Path("extractor.yaml")), ... model=ModelConfig(model="claude-sonnet-4-20250514"), ... workspace=ws, ... runtime_context={"output_schema": Items}, ... ) run = await pes.run("Extract items from data/source.md")

`postprocess_phase_result(phase, result, run)` `async`

Summarize phase: read output.json and validate against output_schema. No-op for other phases.

`validate_phase_outputs(phase, phase_cfg, result, run)` `async`

Execute phase: extend required_outputs validation with plan→findings coverage check.

Bases: BasePES

Audit a document against a checklist of checkpoints.

Before calling run(), place a JSON file at workspace.data_dir / "checkpoints.json" containing a list of checkpoint objects: [{"id": "CP001", "description": "..."}, ...].

Parameters:

Name	Type	Description	Default
`config`	`PESConfig`	PES configuration (use `load_pes_config("auditor.yaml")`).	required
`model`	`ModelConfig`	LLM provider configuration.	required
`workspace`	`WorkspaceHandle`	Isolated workspace for this run.	required
`runtime_context`	`dict[str, Any] \| None`	Must include: - `output_schema` (`type[BaseModel]`): Pydantic model for audit output.	`None`

Phase contracts

plan → working/plan.json (reads data/checkpoints.json)
execute → working/findings/<cp_id>.json per checkpoint
summarize → working/output.json with findings + summary

Example

from scrivai import AuditorPES, ModelConfig, load_pes_config pes = AuditorPES( ... config=load_pes_config(Path("auditor.yaml")), ... model=ModelConfig(model="claude-sonnet-4-20250514"), ... workspace=ws, ... runtime_context={"output_schema": AuditOutput}, ... ) run = await pes.run("Audit data/document.md against checkpoints")

`postprocess_phase_result(phase, result, run)` `async`

Summarize phase: schema validation + verdict/evidence rules.

`validate_phase_outputs(phase, phase_cfg, result, run)` `async`

Execute phase: verify that checkpoint IDs in data/checkpoints.json are covered by findings.

Bases: BasePES

Generate a document by filling a docxtpl template.

The LLM plans which template placeholders to fill, executes data gathering for each placeholder, and summarizes results into a context dict. Optionally renders the final .docx file.

Parameters:

Name	Type	Description	Default
`config`	`PESConfig`	PES configuration (use `load_pes_config("generator.yaml")`).	required
`model`	`ModelConfig`	LLM provider configuration.	required
`workspace`	`WorkspaceHandle`	Isolated workspace for this run.	required
`runtime_context`	`dict[str, Any] \| None`	Must include: - `template_path` (`Path`): Path to docxtpl template. - `context_schema` (`type[BaseModel]`): Schema for template context. - `auto_render` (`bool`, optional): If True, render final.docx.	`None`

Phase contracts

plan → working/plan.json with placeholder fill plan
execute → working/findings/<placeholder>.json per fill
summarize → working/output.json with context dict; output/final.docx if auto_render=True

Example

from scrivai import GeneratorPES, ModelConfig, load_pes_config pes = GeneratorPES( ... config=load_pes_config(Path("generator.yaml")), ... model=ModelConfig(model="claude-sonnet-4-20250514"), ... workspace=ws, ... runtime_context={"template_path": Path("t.docx"), ...}, ... ) run = await pes.run("Fill the project report template")

`build_execution_context(phase, run)` `async`

Plan phase: parse template_path placeholders and inject context['placeholders'].

`postprocess_phase_result(phase, result, run)` `async`

Summarize phase: validate context_schema; render docx if auto_render=True.

`validate_phase_outputs(phase, phase_cfg, result, run)` `async`

Plan phase: verify plan.json covers all placeholders; execute phase: verify findings/ coverage.

Assemble prompts from Jinja2 templates, YAML spec, and markdown fragments.

`init(template_dir, fragments_dir, spec_path)`

Initialize PromptManager.

Parameters:

Name	Type	Description	Default
`template_dir`	`Path`	Directory containing Jinja2 `.j2` template files.	required
`fragments_dir`	`Path`	Directory containing static `.md` prompt fragments.	required
`spec_path`	`Path`	Path to the YAML prompt spec file.	required

Raises:

Type	Description
`FileNotFoundError`	If spec_path does not exist.
`ValueError`	If the YAML structure is invalid.

`build_prompt(operation, phase, context)`

Build a complete prompt by validating context, loading fragments, and rendering the template.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation type (e.g. `"extractor"`).	required
`phase`	`str`	Phase name (e.g. `"plan"`).	required
`context`	`dict[str, Any]`	Render context dict; must satisfy `required_context` from spec.	required

Returns:

Type	Description
`str`	Rendered prompt string.

Raises:

Type	Description
`ValueError`	If required context fields are missing/None or template key is unknown.
`TemplateNotFound`	If the `.j2` file does not exist.

`get_template_spec(operation, phase)`

Read the spec entry for a given operation/phase combination.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation type.	required
`phase`	`str`	Phase name.	required

Returns:

Type	Description
`dict[str, Any]`	Spec dict with `required_context`, `static_fragments`, `artifacts` keys.

Raises:

Type	Description
`ValueError`	If the template key is not defined in the spec.

`validate_context(template_key, template_spec, context)`

Validate that all required context fields are present and non-None.

Parameters:

Name	Type	Description	Default
`template_key`	`str`	Template key for error messages.	required
`template_spec`	`dict[str, Any]`	Spec dict containing `required_context`.	required
`context`	`dict[str, Any]`	Render context to validate.	required

Raises:

Type	Description
`ValueError`	If any required field is missing or None.

`load_fragment(fragment_name)`

Load a static .md prompt fragment by name.

Parameters:

Name	Type	Description	Default
`fragment_name`	`str`	Fragment name, with or without `.md` suffix.	required

Returns:

Type	Description
`str`	Stripped text content of the fragment file.

Raises:

Type	Description
`FileNotFoundError`	If the fragment file does not exist.

PES API Reference

run(task_prompt) async

plan(run, task_prompt) async

execute_phase(run, task_prompt) async

summarize(run, task_prompt) async

build_execution_context(phase, run) async

build_phase_prompt(phase, phase_cfg, context, task_prompt) async

postprocess_phase_result(phase, result, run) async

validate_phase_outputs(phase, phase_cfg, result, run) async

postprocess_phase_result(phase, result, run) async

validate_phase_outputs(phase, phase_cfg, result, run) async

postprocess_phase_result(phase, result, run) async

validate_phase_outputs(phase, phase_cfg, result, run) async

build_execution_context(phase, run) async

postprocess_phase_result(phase, result, run) async

validate_phase_outputs(phase, phase_cfg, result, run) async

__init__(template_dir, fragments_dir, spec_path)

build_prompt(operation, phase, context)

get_template_spec(operation, phase)

validate_context(template_key, template_spec, context)

load_fragment(fragment_name)

`run(task_prompt)` `async`

`plan(run, task_prompt)` `async`

`execute_phase(run, task_prompt)` `async`

`summarize(run, task_prompt)` `async`

`build_execution_context(phase, run)` `async`

`build_phase_prompt(phase, phase_cfg, context, task_prompt)` `async`

`postprocess_phase_result(phase, result, run)` `async`

`validate_phase_outputs(phase, phase_cfg, result, run)` `async`

`postprocess_phase_result(phase, result, run)` `async`

`validate_phase_outputs(phase, phase_cfg, result, run)` `async`

`postprocess_phase_result(phase, result, run)` `async`

`validate_phase_outputs(phase, phase_cfg, result, run)` `async`

`build_execution_context(phase, run)` `async`

`postprocess_phase_result(phase, result, run)` `async`

`validate_phase_outputs(phase, phase_cfg, result, run)` `async`

`init(template_dir, fragments_dir, spec_path)`

`build_prompt(operation, phase, context)`

`get_template_spec(operation, phase)`

`validate_context(template_key, template_spec, context)`

`load_fragment(fragment_name)`