ai-flow-python/app/service/spec/workflow_loader.py

import json
import os

from app.service.spec.step import Step
from app.service.spec.workflow import Workflow


def load_from_directory(template_dir: str) -> Workflow:
    content_path = os.path.join(template_dir, "content.json")
    if not os.path.isfile(content_path):
        raise FileNotFoundError(f"content.json not found in {template_dir}")

    with open(content_path, "r", encoding="utf-8") as f:
        root = json.load(f)

    workflow = _parse_workflow(root)
    workflow.template_path = template_dir
    return workflow


def _parse_workflow(root: dict) -> Workflow:
    flow = Workflow(
        version=root.get("version", ""),
        inputs=root.get("inputs") or {},
        outputs=root.get("outputs") or {},
        order=root.get("order") or [],
    )

    steps: dict[str, Step] = {}
    for step_id in flow.order:
        node = root.get(step_id)
        if node is None:
            continue
        step = Step(
            id=step_id,
            type=node.get("type", ""),
            resources=node.get("resources", ""),
            inputs=node.get("inputs") or {},
            outputs=node.get("outputs") or {},
            retry_count=node.get("retryCount", 0),
        )
        steps[step_id] = step

    flow.steps = steps
    return flow