hip-0013

HIP-13: Workflow Execution Standard. Status Draft. Hanzo's own standard — read this before implementing against it.

HIP-0013: Workflow Execution Standard

Abstract

This proposal defines the Workflow Execution Standard for Hanzo Flow, a visual DAG-based AI workflow builder forked from Langflow. Flow enables AI engineers to design, test, and deploy LangChain-based pipelines through a drag-and-drop canvas, then export production-ready Python or TypeScript artifacts. All workflow systems in the Hanzo ecosystem MUST implement the interfaces defined in this specification.

Repository: github.com/hanzoai/flow Production: flow.hanzo.ai PyPI: hanzoai-flow

Motivation

AI workflow development today suffers from three fundamental problems:

  1. Code-only tooling excludes non-coders. LangChain is powerful but requires fluency in Python, prompt engineering patterns, and async programming. Product managers, domain experts, and junior engineers cannot participate in pipeline design.
  1. No single pane of glass. Teams use scattered notebooks, scripts, and ad-hoc APIs. There is no canonical representation of an AI pipeline that can be versioned, diffed, reviewed, and rolled back like source code.
  1. Provider lock-in. Most workflow tools call LLM providers directly. Switching models means rewriting integration code. There is no uniform abstraction for model selection, billing, and observability.
  1. No Hanzo integration. Existing tools lack first-class support for the Hanzo LLM Gateway (HIP-1), MCP tools (HIP-10), IAM authentication (HIP-26), and credit-based billing (HIP-18).
  1. Workflow vs. automation confusion. Teams conflate AI pipeline construction (prompt chains, RAG, agent loops) with business process automation (webhooks, schedules, CRM integrations). These are distinct concerns requiring distinct tools.

Hanzo Flow solves all five by providing a visual DAG builder backed by LangChain, integrated with Hanzo infrastructure, and clearly scoped to AI pipeline construction.

Specification

LLM:
  - ChatModel          # Conversational LLM (chat completions)
  - CompletionModel    # Text completion LLM
  - ZenModelSelector   # Zen model family picker (Hanzo-specific)
  - ModelRouter        # Route to different models based on input characteristics
Embeddings:
  - OpenAIEmbeddings     # Via Gateway
  - HuggingFaceEmbeddings
  - CohereEmbeddings     # Via Gateway
  - CustomEmbeddings     # User-provided embedding function
VectorStores:
  - Pinecone
  - Weaviate
  - Chroma
  - pgvector          # PostgreSQL with pgvector extension
  - FAISS             # In-memory, for prototyping
  - Qdrant
Agents:
  - ReActAgent         # Reasoning + acting loop
  - PlanAndExecute     # Planning then sequential execution
  - OpenAIFunctions    # Function-calling agent
  - HanzoAgent         # Agent SDK (HIP-9) integration
  - ConversationalAgent
Tools:
  - MCPTool            # Any Hanzo MCP tool (Hanzo-specific)
  - SearchTool         # Web search
  - CalculatorTool
  - PythonREPL         # Sandboxed Python execution
  - APITool            # Call external APIs
  - SQLTool            # Query databases
  - BrowserTool        # Web browsing
Chains:
  - LLMChain           # Prompt template + LLM
  - SequentialChain    # Ordered chain of sub-chains
  - RouterChain        # Route input to different chains
  - ConversationChain  # Chat with memory
  - RetrievalQA        # RAG: retrieve then answer
  - MapReduceChain     # Process documents in parallel then combine
  - SummarizeChain     # Summarize long documents
Memory:
  - ConversationBuffer    # Full conversation history
  - ConversationSummary   # Summarized history
  - ConversationWindow    # Last N messages
  - EntityMemory          # Track entities mentioned
  - VectorStoreMemory     # Similarity-based recall
OutputParsers:
  - JSONParser           # Parse JSON from LLM output
  - PydanticParser       # Parse into Pydantic models
  - ListParser           # Parse comma/newline separated lists
  - RegexParser          # Extract via regex patterns
  - StructuredOutputParser
IO:
  - TextInput            # User text input
  - FileInput            # Upload documents
  - ChatInput            # Conversational input
  - TextOutput           # Display text
  - ChatOutput           # Conversational output
  - WebhookInput         # Receive external triggers
  - WebhookOutput        # Send results externally
            openai_api_base=self.gateway_url,
            temperature=self.temperature,
            max_tokens=self.max_tokens,
        )

MCP Tool Node

Dynamically loads any of Hanzo's 260+ MCP tools as a Flow component.

class MCPToolNode(Component):
    """
    Wraps a Hanzo MCP tool as a Flow node.
    Tool schemas are loaded dynamically from the MCP registry.
    """
    display_name = "MCP Tool"

    inputs = [
        DropdownInput(name="tool_name", display_name="Tool",
                      options_loader=load_mcp_tools,
                      info="Select from 260+ MCP tools"),
        DictInput(name="tool_params", display_name="Parameters",
                  info="Tool-specific parameters (auto-populated from schema)"),
    ]

    def build(self) -> BaseTool:
        schema = mcp_registry.get_schema(self.tool_name)
        return MCPLangChainTool(
            name=self.tool_name,
            description=schema.description,
            args_schema=schema.to_pydantic(),
            mcp_client=self.get_mcp_client(),
        )

Zen Model Selector

Provides a rich model picker showing the Zen model family with metadata.

class ZenModelSelector(Component):
    """
    Model selector showing Zen family with parameter counts,
    context windows, pricing per 1M tokens, and capability tags.
    """
    display_name = "Zen Model Selector"

    inputs = [
        DropdownInput(name="model", display_name="Zen Model",
                      options=[
                          "zen-1b   | 1B params   | 32K ctx  | $0.10/1M",
                          "zen-8b   | 8B params   | 128K ctx | $0.30/1M",
                          "zen-72b  | 72B params  | 128K ctx | $1.00/1M",
                          "zen-480b | 480B params | 256K ctx | $5.00/1M",
                      ]),
        FloatInput(name="temperature", display_name="Temperature", value=0.7),
    ]

    def build(self) -> ChatOpenAI:
        model_id = self.model.split("|")[0].strip()
        return ChatOpenAI(
            model=model_id,
            openai_api_key=self.get_iam_api_key(),
            openai_api_base="https://llm.hanzo.ai",
            temperature=self.temperature,
        )

Credit Billing Hook

Intercepts flow execution to meter token usage and debit credits.

class CreditBillingHook(Component):
    """
    Meters LLM token usage during flow execution and
    creates billing transactions against the user's Hanzo credit balance.
    """
    display_name = "Credit Billing"

    inputs = [
        StrInput(name="org_id", display_name="Organization",
                 info="Hanzo organization to bill"),
        BoolInput(name="enforce_balance", display_name="Enforce Balance",
                  value=True, info="Reject execution if balance insufficient"),
    ]

    def build(self) -> CallbackHandler:
        return BillingCallbackHandler(
            iam_client=self.get_iam_client(),
            org_id=self.org_id,
            enforce_balance=self.enforce_balance,
        )

Flow Definition Schema

A flow is a JSON document describing a directed acyclic graph of components.

{
  "id": "uuid-v4",
  "name": "Customer Support RAG",
  "description": "RAG pipeline for answering customer questions",
  "version": 3,
  "created_at": "2025-06-15T10:30:00Z",
  "updated_at": "2025-07-01T14:22:00Z",
  "created_by": "user-uuid",
  "org_id": "hanzo",
  "tags": ["rag", "support", "production"],
  "nodes": [
    {
      "id": "node-1",
      "type": "TextInput",
      "position": {"x": 100, "y": 200},
      "data": {
        "input_value": "",
        "display_name": "User Question"
      }
    },
    {
      "id": "node-2",
      "type": "OpenAIEmbeddings",
      "position": {"x": 300, "y": 200},
      "data": {
        "model": "text-embedding-3-small"
      }
    },
    {
      "id": "node-3",
      "type": "pgvector",
      "position": {"x": 500, "y": 200},
      "data": {
        "connection_string": "${secrets.PG_VECTOR_URL}",
        "collection_name": "support_docs",
        "search_type": "similarity",
        "k": 5
      }
    },
    {
      "id": "node-4",
      "type": "LLMGatewayNode",
      "position": {"x": 500, "y": 400},
      "data": {
        "model": "zen-72b",
        "temperature": 0.3,
        "max_tokens": 2048
      }
    },
    {
      "id": "node-5",
      "type": "RetrievalQA",
      "position": {"x": 700, "y": 300},
      "data": {
        "chain_type": "stuff",
        "return_source_documents": true
      }
    }
  ],
  "edges": [
    {"source": "node-1", "target": "node-2", "sourceHandle": "text", "targetHandle": "input"},
    {"source": "node-2", "target": "node-3", "sourceHandle": "embeddings", "targetHandle": "embeddings"},
    {"source": "node-1", "target": "node-5", "sourceHandle": "text", "targetHandle": "query"},
    {"source": "node-3", "target": "node-5", "sourceHandle": "retriever", "targetHandle": "retriever"},
    {"source": "node-4", "target": "node-5", "sourceHandle": "llm", "targetHandle": "llm"}
  ]
}

DAG Execution Engine

The execution engine processes flows as directed acyclic graphs.

Topological sort. Before execution, the engine computes a topological ordering of nodes. Nodes with no incoming edges execute first. A node executes only after all its dependencies have completed.

Parallel execution. Independent branches (nodes with no shared ancestry in the current frontier) execute concurrently. The engine uses Python's asyncio.gather to run independent branches in parallel, bounded by a configurable concurrency limit.

Streaming support. LLM nodes support streaming output via Server-Sent Events (SSE). When a flow contains a streaming-capable LLM node connected to an output node, the API returns a streaming response. Intermediate nodes buffer their output; only the final output node streams to the client.

Error propagation. If a node fails, the engine marks all downstream nodes as skipped and returns a partial result containing outputs from all nodes that completed successfully, plus the error from the failed node.

class DAGExecutor:
    """Executes a flow DAG with parallel branch support."""

    async def execute(self, flow: Flow, inputs: dict) -> ExecutionResult:
        graph = self.build_graph(flow)
        order = topological_sort(graph)
        levels = self.compute_levels(order, graph)

        outputs = {}
        for level in levels:
            # All nodes in a level are independent -- execute in parallel
            tasks = [
                self.execute_node(node, outputs, inputs)
                for node in level
            ]
            results = await asyncio.gather(*tasks, return_exceptions=True)

            for node, result in zip(level, results):
                if isinstance(result, Exception):
                    return ExecutionResult(
                        status="failed",
                        outputs=outputs,
                        error={"node": node.id, "message": str(result)},
                    )
                outputs[node.id] = result

        return ExecutionResult(status="completed", outputs=outputs)

API Specification

All API endpoints require OAuth2 bearer tokens from Hanzo IAM.

Flow CRUD

GET    /api/v1/flows                    # List flows (paginated, filterable)
POST   /api/v1/flows                    # Create flow
GET    /api/v1/flows/{flow_id}          # Get flow by ID
PUT    /api/v1/flows/{flow_id}          # Update flow
DELETE /api/v1/flows/{flow_id}          # Delete flow
GET    /api/v1/flows/{flow_id}/versions # List flow versions
GET    /api/v1/flows/{flow_id}/versions/{version} # Get specific version
POST   /api/v1/flows/{flow_id}/rollback/{version} # Rollback to version

Flow Execution

POST   /api/v1/run/{flow_id}           # Execute flow (sync)
POST   /api/v1/run/{flow_id}/stream    # Execute flow (streaming SSE)
POST   /api/v1/run/{flow_id}/async     # Execute flow (async, returns job ID)
GET    /api/v1/jobs/{job_id}           # Poll async job status
GET    /api/v1/jobs/{job_id}/result    # Get async job result
DELETE /api/v1/jobs/{job_id}           # Cancel async job

Templates

GET    /api/v1/templates               # List marketplace templates
GET    /api/v1/templates/{template_id} # Get template
POST   /api/v1/templates               # Publish flow as template
POST   /api/v1/flows/from-template/{template_id} # Create flow from template

Components

GET    /api/v1/components              # List available component types
GET    /api/v1/components/{type}       # Get component schema
GET    /api/v1/components/mcp          # List available MCP tools as components

Execution Request Format

POST /api/v1/run/{flow_id}
{
  "inputs": {
    "text_input": "What is our refund policy?"
  },
  "tweaks": {
    "node-4": {
      "temperature": 0.1,
      "max_tokens": 1024
    }
  },
  "session_id": "optional-session-for-memory"
}

Execution Response Format

{
  "id": "exec-uuid",
  "flow_id": "flow-uuid",
  "status": "completed",
  "duration_ms": 2340,
  "outputs": {
    "node-5": {
      "result": "Our refund policy allows returns within 30 days...",
      "source_documents": [
        {"page_content": "...", "metadata": {"source": "refund-policy.pdf"}}
      ]
    }
  },
  "usage": {
    "total_tokens": 1847,
    "prompt_tokens": 1203,
    "completion_tokens": 644,
    "cost_credits": 0.0018,
    "model": "zen-72b"
  }
}

Versioning

Flow definitions are versioned using an append-only snapshot model.

Template Marketplace

Pre-built flows for common patterns are available in the template marketplace.

| Template | Description | Components Used | |----------|-------------|-----------------| | RAG Chatbot | Conversational RAG with memory | TextInput, Embeddings, pgvector, RetrievalQA, ConversationBuffer | | Document Summarizer | Summarize uploaded documents | FileInput, SummarizeChain, MapReduce, TextOutput | | Code Assistant | Code generation with context | ChatInput, LLMGateway, PythonREPL, ChatOutput | | Classification Pipeline | Classify text into categories | TextInput, LLMChain, JSONParser, WebhookOutput | | Multi-Agent Research | Collaborative research agents | ChatInput, HanzoAgent, SearchTool, BrowserTool, ChatOutput | | Data Extraction | Extract structured data from text | FileInput, LLMChain, PydanticParser, APITool |

Code Export

Flows export to production-ready Python or TypeScript.

Python Export

# Exported from Hanzo Flow: "Customer Support RAG"
# Version: 3 | Exported: 2025-07-01T14:30:00Z

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import PGVector
from langchain.chains import RetrievalQA

# LLM via Hanzo Gateway
llm = ChatOpenAI(
    model="zen-72b",
    openai_api_base="https://llm.hanzo.ai",
    openai_api_key=os.environ["HANZO_API_KEY"],
    temperature=0.3,
    max_tokens=2048,
)

# Embeddings via Hanzo Gateway
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_base="https://llm.hanzo.ai",
    openai_api_key=os.environ["HANZO_API_KEY"],
)

# Vector store
vectorstore = PGVector(
    connection_string=os.environ["PG_VECTOR_URL"],
    collection_name="support_docs",
    embedding_function=embeddings,
)

# RAG chain
chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True,
)

def run(query: str) -> dict:
    return chain.invoke({"query": query})

if __name__ == "__main__":
    import sys
    result = run(sys.argv[1])
    print(result["result"])

TypeScript Export

// Exported from Hanzo Flow: "Customer Support RAG"
// Version: 3 | Exported: 2025-07-01T14:30:00Z

import { ChatOpenAI, OpenAIEmbeddings } from "@langchain/openai";
import { PGVectorStore } from "@langchain/community/vectorstores/pgvector";
import { RetrievalQAChain } from "langchain/chains";

const llm = new ChatOpenAI({
  modelName: "zen-72b",
  configuration: { baseURL: "https://llm.hanzo.ai" },
  openAIApiKey: process.env.HANZO_API_KEY!,
  temperature: 0.3,
  maxTokens: 2048,
});

const embeddings = new OpenAIEmbeddings({
  modelName: "text-embedding-3-small",
  configuration: { baseURL: "https://llm.hanzo.ai" },
  openAIApiKey: process.env.HANZO_API_KEY!,
});

const vectorstore = await PGVectorStore.initialize(embeddings, {
  postgresConnectionOptions: { connectionString: process.env.PG_VECTOR_URL! },
  tableName: "support_docs",
});

const chain = RetrievalQAChain.fromLLM(llm, vectorstore.asRetriever({ k: 5 }));

export async function run(query: string): Promise<string> {
  const result = await chain.invoke({ query });
  return result.text;
}

Implementation

Architecture

                        ┌──────────────────────────────┐
                        │       flow.hanzo.ai           │
                        ├───────────────┬──────────────┤
                        │  React + RF   │  FastAPI     │
                        │  (Canvas UI)  │  (Backend)   │
                        ├───────────────┴──────────────┤
                        │  SQL  │  KV                  │
                        │  (flows,     │  (sessions,   │
                        │   versions)  │   job queue)  │
                        └──────┬───────┴──────┬────────┘
                               │              │
                ┌──────────────┴──┐    ┌──────┴──────────┐
                │  llm.hanzo.ai   │    │  hanzo.id       │
                │  (LLM Gateway)  │    │  (IAM / OAuth)  │
                └─────────────────┘    └─────────────────┘

Production Deployment

Frontend

Backend

Database Schema

-- Flow definitions
CREATE TABLE flows (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    description TEXT,
    org_id VARCHAR(128) NOT NULL,
    created_by UUID NOT NULL,
    is_public BOOLEAN DEFAULT FALSE,
    tags TEXT[],
    current_version INT DEFAULT 1,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

-- Flow version snapshots
CREATE TABLE flow_versions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    flow_id UUID REFERENCES flows(id) ON DELETE CASCADE,
    version INT NOT NULL,
    data JSONB NOT NULL,           -- Complete flow JSON
    description VARCHAR(500),
    created_by UUID NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE (flow_id, version)
);

-- Execution history
CREATE TABLE flow_executions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    flow_id UUID REFERENCES flows(id),
    flow_version INT NOT NULL,
    status VARCHAR(20) NOT NULL,    -- pending, running, completed, failed
    inputs JSONB,
    outputs JSONB,
    error JSONB,
    usage JSONB,                    -- token counts, cost
    duration_ms INT,
    session_id VARCHAR(128),
    executed_by UUID NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Template marketplace
CREATE TABLE flow_templates (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    description TEXT,
    category VARCHAR(64),
    data JSONB NOT NULL,
    author_id UUID NOT NULL,
    org_id VARCHAR(128),
    is_official BOOLEAN DEFAULT FALSE,
    install_count INT DEFAULT 0,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_flows_org ON flows(org_id);
CREATE INDEX idx_executions_flow ON flow_executions(flow_id, created_at DESC);
CREATE INDEX idx_templates_category ON flow_templates(category);

Authentication

All API requests require a valid OAuth2 bearer token from Hanzo IAM (hanzo.id).

from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer

security = HTTPBearer()

async def get_current_user(token = Depends(security)):
    user = await iam_client.validate_token(token.credentials)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

async def check_flow_access(flow_id: str, user = Depends(get_current_user)):
    flow = await get_flow(flow_id)
    if flow.org_id not in user.organizations:
        raise HTTPException(status_code=403, detail="Access denied")
    return flow

Execution Workers

Long-running flows execute on dedicated Celery workers to avoid blocking the API.

from celery import Celery

app = Celery("flow", broker=os.environ["REDIS_URL"])

@app.task(bind=True, max_retries=1, time_limit=300)
def execute_flow_task(self, flow_id: str, version: int, inputs: dict, user_id: str):
    """Execute a flow asynchronously on a worker."""
    flow = load_flow(flow_id, version)
    executor = DAGExecutor(
        billing_hook=CreditBillingHook(user_id=user_id),
        timeout=300,
    )
    result = asyncio.run(executor.execute(flow, inputs))
    store_execution_result(flow_id, version, inputs, result, user_id)
    return result.to_dict()

Security

Sandboxed Code Execution

Custom Python components (PythonREPL, custom nodes) execute in sandboxed environments.

API Key Scoping

Each flow can be assigned a scoped API key for external invocation.

{
  "key_id": "flow_key_xxxx",
  "flow_id": "flow-uuid",
  "permissions": ["execute"],
  "rate_limit": "100/hour",
  "ip_whitelist": ["203.0.113.0/24"],
  "expires_at": "2026-01-01T00:00:00Z"
}

Rate Limiting

Flow execution is rate-limited at three levels:

  1. Per-user: 100 executions per hour (configurable per plan).
  2. Per-flow: 1000 executions per hour (configurable per flow).
  3. Per-organization: 10,000 executions per hour (configurable per org).

Rate limits are enforced in KV using sliding window counters. When a limit is exceeded, the API returns 429 Too Many Requests with a Retry-After header.

Audit Logging

Every flow execution produces an immutable audit log entry.

{
  "event": "flow.executed",
  "flow_id": "flow-uuid",
  "flow_version": 3,
  "user_id": "user-uuid",
  "org_id": "hanzo",
  "ip_address": "203.0.113.42",
  "status": "completed",
  "duration_ms": 2340,
  "tokens_used": 1847,
  "credits_charged": 0.0018,
  "timestamp": "2025-07-01T14:30:00Z"
}

Audit logs are:

Secret Management

Flow definitions MUST NOT contain plaintext secrets. All secrets are referenced using ${secrets.KEY_NAME} syntax and resolved at execution time from Hanzo KMS (HIP-27).

{
  "type": "pgvector",
  "data": {
    "connection_string": "${secrets.PG_VECTOR_URL}"
  }
}

The execution engine resolves secrets by calling KMS with the user's IAM token. Secrets are injected into node configurations in memory and never persisted to the flow definition or execution logs.

Interaction With Other HIPs

| HIP | Relationship | |-----|--------------| | HIP-1 (LLM Gateway) | All LLM calls route through the Gateway. Flow never calls providers directly. | | HIP-9 (Agent SDK) | Flow can embed Agent SDK agents as nodes. Exported code can be deployed as Agent SDK components. | | HIP-10 (MCP) | MCP tools appear as first-class Flow nodes via dynamic schema loading. | | HIP-18 (Payments) | Credit billing hooks meter token usage and debit user balances. | | HIP-26 (IAM) | OAuth2/OIDC authentication for all API access. Organization-scoped flow ownership. | | HIP-27 (Secrets) | All secrets in flow definitions resolved from KMS at execution time. | | HIP-34 (Auto) | Auto can trigger Flow executions as steps in business process automations. Flow does not replace Auto. |

Copyright and related rights waived via CC0.