Skip to content

Agent Harness API Reference

A Python package implementing a compositional execution-state management system for long-horizon agentic code generation loops, with full dependency tracking, feedback-driven patch generation, and selective re-execution.

Package: agent_harness
Python Version: 3.8+
Dependencies: networkx, asttokens, astunparse, openai, pydantic, pytest

Module: dependency_analyzer

Analyzes code dependencies by extracting variable reads and writes, and inferring data flow edges between execution steps.

get_reads(source: str) -> set[str]

Extracts the set of variable names that are read (referenced) in the given source code.

Parameters: - source (str): Python source code as a string

Returns: - set[str]: Set of variable names that are read in the code

Example: 

from agent_harness.dependency_analyzer import get_reads

code = "result = x + y"
reads = get_reads(code)
print(reads)  # {'x', 'y'}

get_writes(source: str) -> set[str]

Extracts the set of variable names that are written (assigned) in the given source code.

Parameters: - source (str): Python source code as a string

Returns: - set[str]: Set of variable names that are assigned in the code

Example: 

from agent_harness.dependency_analyzer import get_writes

code = "result = x + y\noutput = result * 2"
writes = get_writes(code)
print(writes)  # {'result', 'output'}

infer_edges(steps: list[tuple[str, str]]) -> list[tuple[str, str]]

Infers data flow dependencies between execution steps by analyzing variable reads and writes.

Parameters: - steps (list[tuple[str, str]]): List of (step_id, source_code) tuples

Returns: - list[tuple[str, str]]: List of (src_step_id, dst_step_id) edges representing data dependencies

Example: 

from agent_harness.dependency_analyzer import infer_edges

steps = [
    ("step1", "x = 10"),
    ("step2", "y = x + 5"),
    ("step3", "z = y * 2")
]
edges = infer_edges(steps)
print(edges)  # [('step1', 'step2'), ('step2', 'step3')]

Module: esg

Manages the Execution State Graph (ESG), a directed acyclic graph representing the execution history, dependencies, and status of code steps.

class StepStatus(enum.Enum)

Enumeration representing the execution status of a step.

Members: - PENDING: Step has not been executed - EXECUTING: Step is currently executing - SUCCESS: Step executed successfully - FAILED: Step execution failed - STALE: Step output is invalidated due to upstream changes

Example: 

from agent_harness.esg import StepStatus

status = StepStatus.SUCCESS
print(status.name)  # 'SUCCESS'
print(status.value)  # 'success'

class ExecutionStateGraph

Manages the execution state graph with methods for adding steps, tracking dependencies, and managing re-execution.

graph(self) -> nx.DiGraph

Returns the underlying NetworkX directed graph representation of the execution state.

Returns: - nx.DiGraph: NetworkX directed graph object

Example: 

from agent_harness.esg import ExecutionStateGraph

esg = ExecutionStateGraph()
esg.add_step("step1", "x = 10")
nx_graph = esg.graph()
print(nx_graph.nodes())

add_step(self, step_id: str, source: str) -> str

Adds a new execution step to the graph with the provided source code.

Parameters: - step_id (str): Unique identifier for the step - source (str): Python source code for the step

Returns: - str: The step_id that was added

Example: 

from agent_harness.esg import ExecutionStateGraph

esg = ExecutionStateGraph()
step_id = esg.add_step("step1", "x = 10")
print(step_id)  # 'step1'

get_node(self, step_id: str) -> dict[str, Any]

Retrieves metadata and state information for a specific step node.

Parameters: - step_id (str): Identifier of the step

Returns: - dict[str, Any]: Dictionary containing node attributes (source, status, output, etc.)

Example: 

from agent_harness.esg import ExecutionStateGraph

esg = ExecutionStateGraph()
esg.add_step("step1", "x = 10")
node = esg.get_node("step1")
print(node['source'])  # 'x = 10'

add_edge(self, src_step_id: str, dst_step_id: str) -> None

Adds a directed edge representing a data dependency from source step to destination step.

Parameters: - src_step_id (str): Source step identifier - dst_step_id (str): Destination step identifier

Returns: - None

Example: 

from agent_harness.esg import ExecutionStateGraph

esg = ExecutionStateGraph()
esg.add_step("step1", "x = 10")
esg.add_step("step2", "y = x + 5")
esg.add_edge("step1", "step2")

record_output(self, step_id: str, output: dict[str, Any], status: StepStatus) -> None

Records the execution output and status for a completed step.

Parameters: - step_id (str): Identifier of the step - output (dict[str, Any]): Dictionary of variable names to output values - status (StepStatus): Execution status of the step

Returns: - None

Example: 

from agent_harness.esg import ExecutionStateGraph, StepStatus

esg = ExecutionStateGraph()
esg.add_step("step1", "x = 10")
esg.record_output("step1", {"x": 10}, StepStatus.SUCCESS)

get_ancestors(self, step_id: str, variable_names: set[str] | None = None) -> list[str]

Retrieves all ancestor steps that contribute to the specified step's inputs, optionally filtered by variable names.

Parameters: - step_id (str): Identifier of the step - variable_names (set[str] | None): Optional set of variable names to filter ancestors by. If None, all ancestors are returned.

Returns: - list[str]: List of ancestor step identifiers

Example: 

from agent_harness.esg import ExecutionStateGraph

esg = ExecutionStateGraph()
esg.add_step("step1", "x = 10")
esg.add_step("step2", "y = x + 5")
esg.add_step("step3", "z = y * 2")
esg.add_edge("step1", "step2")
esg.add_edge("step2", "step3")

ancestors = esg.get_ancestors("step3")
print(ancestors)  # ['step1', 'step2']

ancestors_filtered = esg.get_ancestors("step3", variable_names={"y"})
print(ancestors_filtered)  # ['step2']

get_subgraph_from(self, step_id: str) -> list[str]

Retrieves all descendant steps that depend (directly or indirectly)