Overview
The GraphSpec class defines the complete structure of an agent graph, including nodes, edges, entry/terminal points, and execution configuration.
Class: GraphSpec
from framework.graph import GraphSpec, NodeSpec, EdgeSpec, EdgeCondition
graph = GraphSpec(
id="calculator-graph",
goal_id="calc-001",
entry_node="input_parser",
terminal_nodes=["output_formatter"],
nodes=[...],
edges=[...]
)
Required Fields
Unique identifier for the graph
ID of the goal this graph achieves
ID of the first node to execute
nodes
list[NodeSpec]
default:"[]"
All node specifications
edges
list[EdgeSpec]
default:"[]"
All edge specifications
Optional Fields
IDs of nodes that end execution
IDs of nodes that pause for human input
entry_points
dict[str, str]
default:"{}"
Named entry points for resuming. Format: {name: node_id}
async_entry_points
list[AsyncEntryPointSpec]
default:"[]"
Asynchronous entry points for concurrent execution streams
Keys available in shared memory
default_model
str
default:"'claude-haiku-4-5-20251001'"
Default LLM model for nodes
Maximum tokens for LLM responses
Maximum node executions before timeout
Maximum retries per node on failure
conversation_mode
str
default:"'continuous'"
Conversation flow mode: “continuous” or “isolated”
Agent-level identity prompt (Layer 1 of onion model)
Human-readable description
Creator: “human” or “builder_agent”
Methods
get_node()
Get a node by ID.
node = graph.get_node("calculator")
if node:
print(node.name)
The node specification, or None if not found
get_outgoing_edges()
Get all edges leaving a node, sorted by priority.
edges = graph.get_outgoing_edges("calculator")
for edge in edges:
print(f"{edge.target} ({edge.condition})")
List of outgoing edges, sorted by priority (highest first)
get_incoming_edges()
Get all edges entering a node.
edges = graph.get_incoming_edges("formatter")
for edge in edges:
print(f"{edge.source} -> {edge.target}")
validate()
Validate the graph structure.
errors = graph.validate()
if errors:
for error in errors:
print(f"Error: {error}")
else:
print("✓ Graph is valid")
List of validation error messages (empty if valid)
detect_fan_out_nodes()
Detect nodes that fan out to multiple targets.
fan_outs = graph.detect_fan_out_nodes()
for source, targets in fan_outs.items():
print(f"{source} -> {targets}")
Mapping of source_node_id to list of parallel target_node_ids
detect_fan_in_nodes()
Detect nodes that receive from multiple sources.
fan_ins = graph.detect_fan_in_nodes()
for target, sources in fan_ins.items():
print(f"{sources} -> {target}")
Mapping of target_node_id to list of source_node_ids
get_entry_point()
Get the appropriate entry point based on session state.
entry_node = graph.get_entry_point(session_state)
print(f"Starting at: {entry_node}")
session_state
dict | None
default:"None"
Optional session state with ‘paused_at’ or ‘resume_from’ key
Node ID to start execution from
Async Entry Points
For multi-entry-point agents that handle concurrent execution streams:
Class: AsyncEntryPointSpec
from framework.graph.edge import AsyncEntryPointSpec
entry = AsyncEntryPointSpec(
id="webhook",
name="Zendesk Webhook Handler",
entry_node="process-webhook",
trigger_type="webhook",
isolation_level="shared",
max_concurrent=10
)
Unique identifier for this entry point
Node ID to start execution from
How this entry point is triggered: webhook, api, timer, event, manual
trigger_config
dict[str, Any]
default:"{}"
Trigger-specific configuration (e.g., webhook URL, timer interval)
State isolation: isolated, shared, or synchronized
Execution priority (higher = more priority)
Maximum concurrent executions for this entry point
Auto-restart on non-fatal failure (0 to disable)
Example: Single Entry Point
from framework.graph import GraphSpec, NodeSpec, EdgeSpec, EdgeCondition
graph = GraphSpec(
id="calculator-graph",
goal_id="calc-001",
entry_node="input_parser",
terminal_nodes=["output_formatter", "error_handler"],
nodes=[
NodeSpec(
id="input_parser",
name="Input Parser",
description="Parse user input",
node_type="event_loop",
input_keys=["expression"],
output_keys=["parsed_expr"],
),
NodeSpec(
id="calculator",
name="Calculator",
description="Perform calculation",
node_type="event_loop",
input_keys=["parsed_expr"],
output_keys=["result"],
),
NodeSpec(
id="output_formatter",
name="Output Formatter",
description="Format the result",
node_type="event_loop",
input_keys=["result"],
output_keys=["formatted_result"],
),
],
edges=[
EdgeSpec(
id="parse-to-calc",
source="input_parser",
target="calculator",
condition=EdgeCondition.ON_SUCCESS,
),
EdgeSpec(
id="calc-to-format",
source="calculator",
target="output_formatter",
condition=EdgeCondition.ON_SUCCESS,
),
],
memory_keys=["expression", "parsed_expr", "result", "formatted_result"],
max_steps=50,
)
# Validate
errors = graph.validate()
if not errors:
print("✓ Graph is valid")
Example: Multi-Entry Point Agent
from framework.graph.edge import AsyncEntryPointSpec
graph = GraphSpec(
id="support-agent-graph",
goal_id="support-001",
entry_node="process-webhook", # Default entry
async_entry_points=[
AsyncEntryPointSpec(
id="webhook",
name="Zendesk Webhook",
entry_node="process-webhook",
trigger_type="webhook",
isolation_level="shared",
max_concurrent=20,
),
AsyncEntryPointSpec(
id="api",
name="API Handler",
entry_node="process-request",
trigger_type="api",
isolation_level="isolated",
max_concurrent=10,
),
],
nodes=[...],
edges=[...],
)