Core API

Flows & Steps

protokol.AbstractFlow

Bases: AbstractStep, ABC

A collection of Steps that form a directed graph. Flows inherit from AbstractStep, meaning an entire Flow can be nested as a single step inside another Flow.

Source code in src/protokol/core/flow.py

class AbstractFlow(AbstractStep, ABC):
    """
    A collection of Steps that form a directed graph.
    Flows inherit from AbstractStep, meaning an entire Flow can be nested as a single step inside another Flow.
    """

    @property
    @abstractmethod
    def steps(self) -> Dict[str, Type[AbstractStep]]:
        """Dictionary mapping step IDs to their corresponding Step classes."""
        pass

    def __init__(self):
        # Instantiate all steps when the flow is created
        self._instances = {
            step_id: step_class()
            for step_id, step_class in self.steps.items()
        }

    def get_step(self, step_id: str) -> AbstractStep:
        """Retrieves an instantiated step by its ID."""
        if step_id not in self._instances:
            raise ValueError(f"Step '{step_id}' not found in flow.")
        return self._instances[step_id]

    def all_steps(self) -> Dict[str, AbstractStep]:
        """Returns all instantiated steps."""
        return self._instances.copy()

    @property
    def start_step_id(self) -> str:
        """Returns the ID of the first step in the flow (used when nesting flows)."""
        return list(self.steps.keys())[0]

    # --- AbstractStep Interface (for Nested Flows) ---
    def get_context(self, plan: RunPlan) -> StepContext:
        """Flows themselves do not have context; the engine descends into their steps."""
        raise NotImplementedError("Engine handles sub-flows directly.")

    def process(self, plan: RunPlan, user_result: Any) -> dict:
        """Flows manipulate global state via their child steps."""
        return {}

    def next(self, plan: RunPlan, output: dict) -> str | None:
        """When a flow finishes, it should return control to the parent flow (or terminate)."""
        return None

start_step_id property

Returns the ID of the first step in the flow (used when nesting flows).

steps abstractmethod property

Dictionary mapping step IDs to their corresponding Step classes.

all_steps()

Returns all instantiated steps.

Source code in src/protokol/core/flow.py

def all_steps(self) -> Dict[str, AbstractStep]:
    """Returns all instantiated steps."""
    return self._instances.copy()

get_context(plan)

Flows themselves do not have context; the engine descends into their steps.

Source code in src/protokol/core/flow.py

def get_context(self, plan: RunPlan) -> StepContext:
    """Flows themselves do not have context; the engine descends into their steps."""
    raise NotImplementedError("Engine handles sub-flows directly.")

get_step(step_id)

Retrieves an instantiated step by its ID.

Source code in src/protokol/core/flow.py

def get_step(self, step_id: str) -> AbstractStep:
    """Retrieves an instantiated step by its ID."""
    if step_id not in self._instances:
        raise ValueError(f"Step '{step_id}' not found in flow.")
    return self._instances[step_id]

next(plan, output)

When a flow finishes, it should return control to the parent flow (or terminate).

Source code in src/protokol/core/flow.py

def next(self, plan: RunPlan, output: dict) -> str | None:
    """When a flow finishes, it should return control to the parent flow (or terminate)."""
    return None

process(plan, user_result)

Flows manipulate global state via their child steps.

Source code in src/protokol/core/flow.py

def process(self, plan: RunPlan, user_result: Any) -> dict:
    """Flows manipulate global state via their child steps."""
    return {}

protokol.AbstractStep

Bases: ABC

A single node in the state machine graph. Steps do NOT execute API calls. They define what is needed, process the result, and decide the next node.

Source code in src/protokol/core/step.py

class AbstractStep(ABC):
    """
    A single node in the state machine graph.
    Steps do NOT execute API calls. They define what is needed, process the result, and decide the next node.
    """

    @property
    @abstractmethod
    def id(self) -> str:
        """The unique identifier for this step within a flow."""
        pass

    @abstractmethod
    def get_context(self, plan: RunPlan) -> StepContext:
        """
        Yields the requirements (prompt, tools) for this step.
        The external engine will execute the LLM based on this context.
        """
        pass

    @abstractmethod
    def process(self, plan: RunPlan, user_result: Any) -> dict:
        """
        Processes the result returned by the user's execution loop (e.g., the LLM response).
        Updates the plan.state as needed.
        Returns a dictionary representing the output of this step.
        """
        pass

    @abstractmethod
    def next(self, plan: RunPlan, output: dict) -> Union[str, List[str], None]:
        """
        Determines the next step(s) in the flow based on the current state and step output.
        Returns:
            - str: ID of the next step.
            - List[str]: IDs of multiple steps to execute in parallel.
            - None: Terminates the flow.
        """
        return None

id abstractmethod property

The unique identifier for this step within a flow.

get_context(plan) abstractmethod

Yields the requirements (prompt, tools) for this step. The external engine will execute the LLM based on this context.

Source code in src/protokol/core/step.py

@abstractmethod
def get_context(self, plan: RunPlan) -> StepContext:
    """
    Yields the requirements (prompt, tools) for this step.
    The external engine will execute the LLM based on this context.
    """
    pass

next(plan, output) abstractmethod

Determines the next step(s) in the flow based on the current state and step output. Returns: - str: ID of the next step. - List[str]: IDs of multiple steps to execute in parallel. - None: Terminates the flow.

Source code in src/protokol/core/step.py

@abstractmethod
def next(self, plan: RunPlan, output: dict) -> Union[str, List[str], None]:
    """
    Determines the next step(s) in the flow based on the current state and step output.
    Returns:
        - str: ID of the next step.
        - List[str]: IDs of multiple steps to execute in parallel.
        - None: Terminates the flow.
    """
    return None

process(plan, user_result) abstractmethod

Processes the result returned by the user's execution loop (e.g., the LLM response). Updates the plan.state as needed. Returns a dictionary representing the output of this step.

Source code in src/protokol/core/step.py

@abstractmethod
def process(self, plan: RunPlan, user_result: Any) -> dict:
    """
    Processes the result returned by the user's execution loop (e.g., the LLM response).
    Updates the plan.state as needed.
    Returns a dictionary representing the output of this step.
    """
    pass

Types

protokol.StepContext dataclass

Yielded by a Step to request execution from the user's domain. It contains the prompt, required tools, and any kwargs (e.g., temperature) the LLM needs.

Source code in src/protokol/core/types.py

@dataclass
class StepContext:
    """
    Yielded by a Step to request execution from the user's domain.
    It contains the prompt, required tools, and any kwargs (e.g., temperature) the LLM needs.
    """
    prompt: str
    tools: List[str] = field(default_factory=list)
    kwargs: Dict[str, Any] = field(default_factory=dict)

protokol.RunPlan dataclass

The complete, serializable memory of the state machine. This object is passed between steps and can be persisted to a database to pause/resume workflows.

Source code in src/protokol/core/types.py

@dataclass
class RunPlan:
    """
    The complete, serializable memory of the state machine.
    This object is passed between steps and can be persisted to a database to pause/resume workflows.
    """
    # Global state payload manipulated by the steps
    state: Dict[str, Any] = field(default_factory=dict)

    # Tracks retries for individual steps
    step_states: Dict[str, StepState] = field(default_factory=dict)

    # Flow control pointers
    current_step: Union[str, List[str], None] = None
    is_terminal: bool = False
    is_waiting: bool = False # Flag indicating the flow is paused waiting for human input
    failed: bool = False
    error: Optional[str] = None

    # Tracks nested flow context (so the engine knows where to return)
    call_stack: List[Dict[str, Any]] = field(default_factory=list)

    # Immutable audit log of every step executed
    trace: List[TraceEntry] = field(default_factory=list)

    def get_step_state(self, step_id: str) -> StepState:
        """Retrieves or initializes the retry state for a specific step."""
        if step_id not in self.step_states:
            self.step_states[step_id] = StepState()
        return self.step_states[step_id]

    def reset_step_state(self, step_id: str) -> None:
        """Resets attempt counters for a step after a successful run."""
        state = self.get_step_state(step_id)
        state.attempt = 0
        state.last_error = None

    def add_trace_entry(
        self,
        *,
        step: str,
        attempt: int,
        result: Any,
        next_step: NextStep,
        parallel: bool,
    ) -> None:
        entry = TraceEntry(step=step, attempt=attempt, result=result, next=next_step, parallel=parallel)
        self.trace.append(entry)

    def to_dict(self) -> Dict[str, Any]:
        return {
            "state": self.state,
            "step_states": {k: {
                "attempt": v.attempt,
                "max_retries": v.max_retries,
                "last_error": v.last_error,
            } for k, v in self.step_states.items()},
            "current_step": self.current_step,
            "is_terminal": self.is_terminal,
            "is_waiting": self.is_waiting,
            "failed": self.failed,
            "error": self.error,
            "call_stack": self.call_stack,
            "trace": [entry.to_dict() for entry in self.trace],
        }

    def to_json(self) -> str:
        """Serializes the RunPlan for database persistence."""
        return json.dumps(self.to_dict())

    @classmethod
    def from_json(cls, data: str) -> 'RunPlan':
        """Hydrates a RunPlan from a JSON string."""
        raw = json.loads(data)
        return cls.from_dict(raw)

    @classmethod
    def from_dict(cls, raw: Dict[str, Any]) -> 'RunPlan':
        if not isinstance(raw, dict):
            raise ValueError("RunPlan data must be a dictionary")

        state = raw.get("state", {})
        if not isinstance(state, dict):
            raise ValueError("RunPlan.state must be a dictionary")

        step_states_raw = raw.get("step_states", {})
        if not isinstance(step_states_raw, dict):
            raise ValueError("RunPlan.step_states must be a dictionary")
        step_states = {k: StepState(**v) for k, v in step_states_raw.items() if isinstance(v, dict)}

        current_step = raw.get("current_step")
        if current_step is not None and not isinstance(current_step, (str, list)):
            raise ValueError("RunPlan.current_step must be a string, list of strings, or None")
        if isinstance(current_step, list):
            for item in current_step:
                if not isinstance(item, str):
                    raise ValueError("RunPlan.current_step list must contain only strings")

        call_stack = raw.get("call_stack", [])
        if not isinstance(call_stack, list):
            raise ValueError("RunPlan.call_stack must be a list")
        for frame in call_stack:
            if not isinstance(frame, dict):
                raise ValueError("RunPlan.call_stack entries must be dictionaries")

        trace_raw = raw.get("trace", [])
        if not isinstance(trace_raw, list):
            raise ValueError("RunPlan.trace must be a list")
        trace = [TraceEntry.from_raw(entry) for entry in trace_raw]

        return cls(
            state=state,
            step_states=step_states,
            current_step=current_step,
            is_terminal=bool(raw.get("is_terminal", False)),
            is_waiting=bool(raw.get("is_waiting", False)),
            failed=bool(raw.get("failed", False)),
            error=raw.get("error"),
            call_stack=call_stack,
            trace=trace,
        )

from_json(data) classmethod

Hydrates a RunPlan from a JSON string.

Source code in src/protokol/core/types.py

@classmethod
def from_json(cls, data: str) -> 'RunPlan':
    """Hydrates a RunPlan from a JSON string."""
    raw = json.loads(data)
    return cls.from_dict(raw)

get_step_state(step_id)

Retrieves or initializes the retry state for a specific step.

Source code in src/protokol/core/types.py

def get_step_state(self, step_id: str) -> StepState:
    """Retrieves or initializes the retry state for a specific step."""
    if step_id not in self.step_states:
        self.step_states[step_id] = StepState()
    return self.step_states[step_id]

reset_step_state(step_id)

Resets attempt counters for a step after a successful run.

Source code in src/protokol/core/types.py

def reset_step_state(self, step_id: str) -> None:
    """Resets attempt counters for a step after a successful run."""
    state = self.get_step_state(step_id)
    state.attempt = 0
    state.last_error = None

to_json()

Serializes the RunPlan for database persistence.

Source code in src/protokol/core/types.py

def to_json(self) -> str:
    """Serializes the RunPlan for database persistence."""
    return json.dumps(self.to_dict())