Workflow API¶

The fcc.workflow package models FCC workflows as directed graphs. Each graph consists of typed nodes (personas, gates) and typed edges (handoff, feedback, gate). Three bundled workflow definitions ship with the package: a 5-node base sequence, a 20-node extended sequence, and a full 24-node complete graph.

flowchart LR
    Load[Load Graph JSON] --> RC[RC: Find]
    RC -->|research_inventory| BC[BC: Create]
    BC -->|blueprints_specs| DE[DE: Critique]
    DE -->|publish_handoff| RB[RB: Create]
    DE -->|publish_handoff| UG[UG: Create]
    RB -.->|operational_feedback| BC
    UG -.->|usability_feedback| BC
    BC -.->|revision| RC
    RB --> Done((Done))
    UG --> Done

Data Models¶

WorkflowMeta¶

Metadata attached to every workflow graph.

from fcc.workflow.models import WorkflowMeta

meta = WorkflowMeta(id="base", title="Base FCC Sequence", description="5-node core workflow")
print(meta.id)     # "base"
print(meta.title)  # "Base FCC Sequence"

WorkflowNode¶

A node in the graph. The type field classifies nodes as persona, gate, start, or end.

from fcc.workflow.models import WorkflowNode

node = WorkflowNode(id="RC", name="Research Crafter", type="persona")
print(node.id)    # "RC"
print(node.type)  # "persona"

WorkflowEdge¶

A directed edge between two nodes. The type field is either handoff (forward flow) or feedback (backward loop).

from fcc.workflow.models import WorkflowEdge

edge = WorkflowEdge(from_id="RC", to_id="BC", label="research_to_blueprint", type="handoff")
print(edge.from_id)  # "RC"
print(edge.to_id)    # "BC"
print(edge.type)     # "handoff"

# Serialize to dict (round-trips through JSON)
print(edge.to_dict())
# {"from": "RC", "to": "BC", "label": "research_to_blueprint", "type": "handoff"}

WorkflowGraph¶

WorkflowGraph is the main entry point. It builds an adjacency list on construction and supports node lookup, edge queries, and graph traversal.

Loading a Graph¶

from fcc._resources import get_workflows_dir, get_schemas_dir
from fcc.workflow.graph import WorkflowGraph

# Load without validation
graph = WorkflowGraph.from_json(get_workflows_dir() / "base_sequence.json")
print(f"Nodes: {len(graph)}, Edges: {len(graph.edges)}")

# Load with JSON Schema validation
graph = WorkflowGraph.from_json_validated(
    json_path=get_workflows_dir() / "base_sequence.json",
    schema_path=get_schemas_dir() / "workflow.schema.json",
)

Inspecting the Graph¶

from fcc._resources import get_workflows_dir
from fcc.workflow.graph import WorkflowGraph

graph = WorkflowGraph.from_json(get_workflows_dir() / "base_sequence.json")

# Metadata
print(graph.meta.id)     # e.g. "base_sequence"
print(graph.meta.title)  # e.g. "Base FCC Sequence"

# List all node IDs
print(graph.node_ids)    # ["RC", "BC", "DE", "RB", "UG"]

# Check membership
print("RC" in graph)  # True
print("XY" in graph)  # False

# Get a node by ID
node = graph.get_node("RC")
print(f"{node.name} ({node.type})")  # "Research Crafter (persona)"

Edge Queries¶

from fcc._resources import get_workflows_dir
from fcc.workflow.graph import WorkflowGraph

graph = WorkflowGraph.from_json(get_workflows_dir() / "base_sequence.json")

# Outgoing edges from a node
for edge in graph.outgoing_edges("RC"):
    print(f"  RC -> {edge.to_id}: {edge.label} ({edge.type})")

# Incoming edges to a node
for edge in graph.incoming_edges("BC"):
    print(f"  {edge.from_id} -> BC: {edge.label} ({edge.type})")

# Successor and predecessor nodes
successors = graph.successors("RC")
print([n.id for n in successors])  # nodes reachable from RC

predecessors = graph.predecessors("DE")
print([n.id for n in predecessors])  # nodes feeding into DE

# Filter edges by type
handoffs = graph.handoffs()
feedbacks = graph.feedbacks()
print(f"Handoffs: {len(handoffs)}, Feedbacks: {len(feedbacks)}")

# Arbitrary edge type
gate_edges = graph.edges_by_type("gate")

Traversal¶

The graph supports topological ordering (using only handoff edges, ignoring feedback loops) and BFS traversal (following all edges).

from fcc._resources import get_workflows_dir
from fcc.workflow.graph import WorkflowGraph

graph = WorkflowGraph.from_json(get_workflows_dir() / "base_sequence.json")

# Topological order (deterministic, feedback edges excluded)
order = graph.topological_order()
print(order)  # e.g. ["RC", "BC", "RB", "UG", "DE"]

# BFS from a starting node (follows all edge types)
bfs = graph.bfs_from("RC")
print(bfs)  # ["RC", "BC", "DE", "RB", "UG", ...]

Serialization¶

from fcc._resources import get_workflows_dir
from fcc.workflow.graph import WorkflowGraph

graph = WorkflowGraph.from_json(get_workflows_dir() / "base_sequence.json")

# To dict (JSON-compatible)
data = graph.to_dict()
print(data["meta"]["title"])

# Write to JSON file
graph.to_json("/tmp/my_workflow.json", indent=2)

Building a Graph Programmatically¶

from fcc.workflow.models import WorkflowMeta, WorkflowNode, WorkflowEdge
from fcc.workflow.graph import WorkflowGraph

meta = WorkflowMeta(id="custom", title="Custom Pipeline")
nodes = [
    WorkflowNode(id="A", name="Analyzer", type="persona"),
    WorkflowNode(id="B", name="Builder", type="persona"),
    WorkflowNode(id="R", name="Reviewer", type="persona"),
]
edges = [
    WorkflowEdge(from_id="A", to_id="B", label="analysis_done", type="handoff"),
    WorkflowEdge(from_id="B", to_id="R", label="build_done", type="handoff"),
    WorkflowEdge(from_id="R", to_id="A", label="revision_needed", type="feedback"),
]

graph = WorkflowGraph(meta=meta, nodes=nodes, edges=edges)
print(graph.topological_order())  # ["A", "B", "R"]
print(graph.bfs_from("A"))        # ["A", "B", "R"]

Loading All Three Bundled Workflows¶

from fcc._resources import get_workflows_dir
from fcc.workflow.graph import WorkflowGraph

workflows_dir = get_workflows_dir()

base = WorkflowGraph.from_json(workflows_dir / "base_sequence.json")
extended = WorkflowGraph.from_json(workflows_dir / "extended_sequence.json")
complete = WorkflowGraph.from_json(workflows_dir / "complete_24.json")

print(f"Base: {len(base)} nodes, {len(base.edges)} edges")
print(f"Extended: {len(extended)} nodes, {len(extended.edges)} edges")
print(f"Complete: {len(complete)} nodes, {len(complete.edges)} edges")