How to Use AWEL (Agentic Workflow Execution Language) for Building Complex Agent Workflows

AWEL (Agentic Workflow Execution Language) is the core declarative orchestration engine in Derisk that transforms JSON-serializable workflow graphs into asynchronous DAGs of operators, enabling type-safe wiring and extensible agent pipelines.

AWEL (Agentic Workflow Execution Language) powers the agent-centric pipelines in the derisk-ai/openderisk repository by letting you define workflows as portable JSON structures and execute them as runnable Directed Acyclic Graphs (DAGs). This architecture decouples workflow design from runtime execution, allowing you to compose complex agent behaviors through declarative graphs while maintaining strict type safety and asynchronous performance.

Understanding the AWEL Architecture

AWEL organizes workflows into four primary layers: the declarative layer (FlowPanel), the factory layer (FlowFactory), the runtime layer (DAG and DAGNode), and the integration layer (triggers and resources).

Core Components

• FlowPanel – The serializable external model defined in packages/derisk-core/src/derisk/core/awel/flow/flow_factory.py that stores flow metadata, UI layout, variables, and the raw graph definition. It serves as the entry point for any user-defined workflow.

• FlowFactory – The builder class in flow_factory.py that converts a FlowPanel into a concrete DAG. It performs topological sorting, resolves operator and resource classes via _get_operator_class, and wires nodes together through build() and build_dag() methods.

• DAG and DAGNode – The execution engine defined in packages/derisk-core/src/derisk/core/awel/dag/base.py. DAG holds the graph structure and schedules tasks, while each operator inherits from DAGNode (via BaseOperator) to execute asynchronously.

• Operators – Reusable computational blocks (LLM calls, branching, mapping) defined in packages/derisk-core/src/derisk/core/awel/operators/base.py. Each operator exposes metadata describing inputs and outputs, enabling the factory to validate connections before runtime.

• Resources – External service connectors (OpenAI clients, databases) declared in packages/derisk-core/src/derisk/core/awel/resource/__init__.py. The factory instantiates these before operator construction, injecting them as dependencies.

• Triggers – Entry points that convert HTTP requests or scheduler events into DAG executions. Implemented in packages/derisk-core/src/derisk/core/awel/trigger/http_trigger.py and managed by TriggerManager.

• Runner – The execution wrapper in packages/derisk-core/src/derisk/core/awel/runner/local_runner.py. DefaultWorkflowRunner executes operators synchronously during development via setup_dev_environment.

Building Your First AWEL Workflow

To create a runnable workflow, you define a FlowPanel containing nodes (operators) and edges (data flow), then pass it to FlowFactory.build().

Hello World Example

The following example creates a minimal two-node pipeline using InputOperator and JoinOperator:

from derisk.core.awel.flow.flow_factory import FlowFactory, FlowPanel, FlowData, FlowNodeData, FlowEdgeData
from derisk.core.awel.flow.base import State
from derisk.core.awel.operators.common_operator import InputOperator, JoinOperator

# Define the nodes

nodes = [
    FlowNodeData(
        id="input_node",
        width=200,
        height=100,
        position={"x": 0, "y": 0, "zoom": 1},
        type="operator",
        data=InputOperator.metadata,
    ),
    FlowNodeData(
        id="join_node",
        width=200,
        height=100,
        position={"x": 300, "y": 0, "zoom": 1},
        type="operator",
        data=JoinOperator.metadata,
    ),
]

# Connect them

edges = [
    FlowEdgeData(
        source="input_node",
        source_order=0,
        target="join_node",
        target_order=0,
        id="e0",
    )
]

# Assemble the FlowPanel

flow_panel = FlowPanel(
    uid="demo-hello",
    label="Hello-World AWEL Flow",
    name="hello_flow",
    flow_category=None,
    flow_data=FlowData(nodes=nodes, edges=edges, viewport={"x": 0, "y": 0, "zoom": 1}),
    state=State.DEVELOPING,
)

# Build and run

factory = FlowFactory()
dag = factory.build(flow_panel)

from derisk.core.awel import setup_dev_environment
setup_dev_environment([dag])

When you call setup_dev_environment(), the function creates a temporary SystemApp context and executes the DAG immediately in your REPL environment.

Adding HTTP Triggers to AWEL Flows

To expose a workflow via REST API, replace the input operator with an HttpTrigger node. This registers a route in the underlying FastAPI application created by setup_dev_environment.

from derisk.core.awel.flow.flow_factory import FlowFactory, FlowPanel, FlowData, FlowNodeData, FlowEdgeData
from derisk.core.awel.trigger.http_trigger import HttpTrigger
from derisk.core.awel.operators.common_operator import InputOperator

trigger_node = FlowNodeData(
    id="http_trigger",
    width=200,
    height=100,
    position={"x": 0, "y": 0, "zoom": 1},
    type="operator",
    data=HttpTrigger.metadata,
)

llm_node = FlowNodeData(
    id="llm_node",
    width=250,
    height=120,
    position={"x": 300, "y": 0, "zoom": 1},
    type="operator",
    data=InputOperator.metadata,
)

edges = [
    FlowEdgeData(
        source="http_trigger",
        source_order=0,
        target="llm_node",
        target_order=0,
        id="e0",
    )
]

flow_panel = FlowPanel(
    uid="http-demo",
    label="HTTP-Triggered LLM Flow",
    name="http_llm_flow",
    flow_data=FlowData(nodes=[trigger_node, llm_node], edges=edges, viewport={"x": 0, "y": 0, "zoom": 1}),
)

dag = FlowFactory().build(flow_panel)

# Start dev server on port 5555

setup_dev_environment([dag], host="127.0.0.1", port=5555)

With the server running, POST requests to /awel/http_trigger automatically instantiate the DAG and stream the request payload through your operator chain.

Integrating External Resources

AWEL supports dependency injection for external services through resource nodes. During FlowFactory.build(), the factory uses import_from_string to instantiate resource classes before wiring them into operators.

OpenAI Client Example

from derisk.core.awel.flow.flow_factory import FlowFactory, FlowPanel, FlowData, FlowNodeData, FlowEdgeData
from derisk.core.awel.operators.common_operator import InputOperator
from derisk.core.awel.resource import ResourceMetadata

# Resource definition

resource_node = FlowNodeData(
    id="openai_client",
    width=150,
    height=80,
    position={"x": -200, "y": 0, "zoom": 1},
    type="resource",
    data=ResourceMetadata(
        label="OpenAI Client",
        description="Provides ChatGPT API access",
        icon=None,
        type_cls="derisk_client.resource.OpenAIClient",
        parameters=[],
    ),
)

# Operator expecting the resource

llm_node = FlowNodeData(
    id="llm_node",
    width=250,
    height=120,
    position={"x": 0, "y": 0, "zoom": 1},
    type="operator",
    data=InputOperator.metadata,
)

edges = [
    FlowEdgeData(
        source="openai_client",
        source_order=0,
        target="llm_node",
        target_order=0,  # Maps to the first resource parameter

        id="e0",
    )
]

flow_panel = FlowPanel(
    uid="resource-demo",
    label="Flow with OpenAI Resource",
    name="resource_flow",
    flow_data=FlowData(nodes=[resource_node, llm_node], edges=edges, viewport={"x": 0, "y": 0, "zoom": 1}),
)

dag = FlowFactory().build(flow_panel)
setup_dev_environment([dag])

The _topological_sort method in flow_factory.py ensures resources instantiate before dependent operators, guaranteeing that the OpenAIClient is available when the LLM operator's run coroutine executes.

AWEL Execution Flow and Runtime

Understanding the runtime path helps debug performance bottlenecks and deployment issues. The execution follows five strict phases:

1. Definition – You create a FlowPanel as JSON or Python objects containing FlowNodeData and FlowEdgeData arrays.

2. Building – FlowFactory.build(flow_panel) validates metadata, performs topological sorting, and constructs the DAG object stored in packages/derisk-core/src/derisk/core/awel/dag/base.py.

3. Registration – The DAGManager (initialized via initialize_awel in packages/derisk-core/src/derisk/core/awel/__init__.py) indexes the DAG for trigger matching.

4. Triggering – An HttpTrigger or scheduler event invokes the TriggerManager, which looks up the corresponding DAG and creates an execution context.

5. Execution – The DefaultWorkflowRunner in local_runner.py walks the DAG, awaiting each operator's run coroutine and propagating outputs along the pre-wired edges.

This pipeline guarantees type-safe wiring through metadata validation at build time, while maintaining asynchronous execution throughout the runtime to prevent I/O blocking during LLM calls or database queries.

Summary

• AWEL workflows are declarative JSON graphs stored in FlowPanel objects and converted to executable DAGs by FlowFactory.
• The architecture separates concerns into operators (logic), resources (dependencies), and triggers (entry points), enabling modular extensibility.
• Use setup_dev_environment() from packages/derisk-core/src/derisk/core/awel/__init__.py for local development; it handles FastAPI registration and SystemApp initialization automatically.
• Resource injection occurs during the build phase via _topological_sort, ensuring external clients are instantiated before operators that depend on them.
• For production deployment, the same FlowPanel JSON can be loaded by derisk serve flow service (see derisk/serve/flow/service/service.py) or the CLI command derisk flow run (see derisk/client/_cli.py).

Frequently Asked Questions

What is the difference between a FlowPanel and a DAG in AWEL?

A FlowPanel is the serializable, static description of your workflow—including node positions, metadata, and edge connections—stored as JSON-friendly Pydantic models in flow_factory.py. A DAG is the runtime execution graph instantiated in dag/base.py that manages asynchronous task scheduling and state. You define a FlowPanel; AWEL converts it into a DAG via FlowFactory.build().

How does AWEL handle type safety between connected operators?

AWEL enforces type safety through metadata validation during the build phase. Each operator class exposes a metadata object (defined in operators/base.py) declaring its input and output schemas. When FlowFactory.build() processes edges, it validates that the source operator's output signature matches the target operator's input requirements before constructing the DAG, preventing runtime type mismatches.

Can I run AWEL workflows without HTTP triggers?

Yes. The setup_dev_environment() function in packages/derisk-core/src/derisk/core/awel/__init__.py creates a standalone SystemApp and DefaultWorkflowRunner that executes DAGs synchronously in your local Python process. This is ideal for unit testing, debugging, or batch processing without exposing network endpoints.

How do I add custom operators to an AWEL workflow?

Create a Python class inheriting from BaseOperator (in operators/base.py) and expose a metadata class attribute describing its inputs, outputs, and parameters. Place the class in your Python path, then reference its fully-qualified name in the type_cls field of a FlowNodeData object. FlowFactory uses import_from_string to dynamically load and instantiate your custom operator during the build process.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}