---
name: agentbyte-workflow
description: Guide for creating deterministic workflows in Agentbyte with Workflow, WorkflowRunner, reusable steps, and event-based observability. Use this when implementing DAG-style execution instead of autonomous orchestration.
license: MIT
---

Use this skill when implementing deterministic multi-step execution graphs in Agentbyte using `Workflow`, `WorkflowRunner`, and workflow steps.

## Inputs to confirm
- Whether the problem is better modeled as a deterministic workflow or an autonomous orchestrator
- Which steps, input/output types, and routing conditions are needed
- Whether the workflow is linear, conditional, or includes fan-out/fan-in
- Whether checkpointing, cancellation, or event-stream observability is required

## When to use workflow vs orchestration
- Use `Workflow` when the execution graph should be explicit and deterministic.
- Use orchestrators when agent-to-agent collaboration should be autonomous or model-directed.
- Prefer workflow when steps are fixed functions, transformations, HTTP calls, or wrapped agents in a known graph.

## Pattern: Basic Workflow

```python
import asyncio

from pydantic import BaseModel

from agentbyte.workflow import (
    FunctionStep,
    StepMetadata,
    Workflow,
    WorkflowConfig,
    WorkflowRunner,
)


class NumberInput(BaseModel):
    value: int


class NumberOutput(BaseModel):
    value: int


async def double(input_data: NumberInput, context) -> NumberOutput:
    return NumberOutput(value=input_data.value * 2)


async def add_ten(input_data: NumberOutput, context) -> NumberOutput:
    return NumberOutput(value=input_data.value + 10)


workflow = Workflow(WorkflowConfig(name="math_pipeline"))
workflow.chain(
    FunctionStep("double", StepMetadata(name="double"), NumberInput, NumberOutput, double),
    FunctionStep("add_ten", StepMetadata(name="add_ten"), NumberOutput, NumberOutput, add_ten),
)

runner = WorkflowRunner()
execution = asyncio.run(runner.run(workflow, {"value": 5}))
print(execution.state["add_ten_output"])
```

## Pattern: Streaming Workflow Events

```python
async for event in WorkflowRunner().run_stream(workflow, {"value": 5}):
    print(type(event).__name__, event)
```

Key event types:
- `WorkflowStartedEvent`
- `StepStartedEvent`
- `StepProgressEvent`
- `StepCompletedEvent`
- `StepFailedEvent`
- `WorkflowCompletedEvent`
- `WorkflowCancelledEvent`

Use `run_stream()` when you need logging, live progress, or execution telemetry. Workflow logging is event-driven today, not middleware-driven.

## Common Step Types
- `FunctionStep`: wrap sync/async Python functions
- `TransformStep`: map fields between schemas
- `HttpStep`: perform HTTP requests
- `AgentbyteAgentStep`: wrap an Agentbyte agent inside a deterministic workflow
- `EchoStep`: simple testing/demo step

## Construction Process
1. Define clear Pydantic input and output models for each step.
2. Choose the right step type for each unit of work.
3. Build the graph with `Workflow(...).chain(...)` for linear flows or `add_step(...)` plus `add_edge(...)` for custom routing.
4. Validate by running through `WorkflowRunner`, which checks workflow structure before execution.
5. Prefer `run_stream()` if you need observability or custom logging.

## Observability And Progress
- `WorkflowCompletedEvent` includes the full `WorkflowExecution`, including overall start/end timestamps.
- `StepCompletedEvent` and `StepFailedEvent` include `duration_seconds`.
- `StepStartedEvent` includes `input_data`.
- `StepCompletedEvent` includes `output_data`.
- `Context.emit_progress(...)` inside a step produces `StepProgressEvent`.
- The runner also creates per-step OTel spans when tracing is enabled.

## Defaults
- Use `Workflow.chain(...)` for simple linear flows.
- Use `WorkflowRunner.run_stream(...)` when debugging or logging execution.
- Use `context.emit_progress(...)` for long-running steps that should report intermediate status.

## Pattern: AgentContext Opt-In with AgentbyteAgentStep

Workflows are pipelines — they do not thread conversation history by default.
When an `AgentbyteAgentStep` needs session continuity (e.g., the same agent is
called across multiple workflow invocations), pass an `AgentContext` via
`agent_context` on the input model.

```python
import asyncio

from agentbyte.context import AgentContext
from agentbyte.session_store import CachedFileSessionStore
from agentbyte.workflow import Workflow, WorkflowConfig, WorkflowRunner
from agentbyte.workflow.steps import AgentbyteAgentStep, AgentbyteAgentInput, StepMetadata

# Retrieve or create a persistent context for the agent in this step
store = CachedFileSessionStore(".sessions")
session_id = "user-abc"
ctx = await store.get(session_id) or AgentContext(session_id=session_id)

# Pass context as part of the step input — it is not coupled to the workflow graph
step = AgentbyteAgentStep("writer", StepMetadata(name="writer"), writer_agent)
workflow = Workflow(WorkflowConfig(name="draft_pipeline"))
workflow.add_step(step)

runner = WorkflowRunner()
execution = await runner.run(
    workflow,
    AgentbyteAgentInput(task="Write about AI trends", agent_context=ctx),
)

# Persist updated context after the workflow completes
await store.save(session_id, ctx)
```

Two distinct context types coexist in a workflow:
| Type | Scope | Purpose |
|------|-------|---------|
| `Context` (workflow) | One workflow run | Pass state between steps |
| `AgentContext` | Across runs | Persist agent conversation history |

`agent_context` is always optional. Omit it (or pass `None`) for a stateless
step that starts fresh every time.

## Guardrails
- Do not use workflow when the control flow should be decided dynamically by an LLM at runtime.
- Do not skip typed input/output models for non-trivial steps.
- Do not assume workflow has agent-style middleware; logging should currently be done from streamed workflow events.
- Prefer checkpoint/resume only when the workflow graph is stable and long-running execution makes it worthwhile.
- Do not thread `AgentContext` through the workflow `Context` dict — keep it in `AgentbyteAgentInput.agent_context`.
