Core Concepts

Flow Execution Lifecycle

Phase	Operations	Validation	Next Phase
Validation	• Schema validation of flow JSON • Dependency analysis and cycle detection • Variable reference validation	All checks must pass	Queue or Reject
Queueing	• Flow added to in-memory execution queue • Assigned unique identifier (timestamp + execution number) • Waits for available execution slot	Concurrency limits	Execute when slot available
Execution	• Flow execution context created with isolated variable registry • Input variables merged with flow variables (input variables take precedence) • Nodes execute sequentially according to dependency order • Variables resolved just-in-time during node execution • Parallel branches handle failures independently	Runtime validation per node	Complete or Fail
Completion	• Temporary files cleaned up • Variable registry cleared • Flow removed from active execution tracking	Resource cleanup verification	Return results

Model Context Protocol (MCP) Integration

The OpenFlow Protocol supports integration with Model Context Protocol (MCP) servers, enabling LLMs to access external tools and data sources in real-time. MCP integration extends LLM capabilities beyond their training data.

MCP Server Configuration

{
  name: "server_identifier (alphanumeric_underscore, unique across all MCP servers, e.g., 'github_api', 'database_tools')",
  url: "mcp_endpoint_url (https_websocket_url, e.g., 'wss://api.example.com/mcp')",
  description: "server_description (human_readable, optional, explains server capabilities)",
  timeout: "request_timeout_ms (integer, 5000-120000, default: 30000)",
  retry_attempts: "max_retry_attempts (integer, 0-5, default: 3)",
  auth: "authentication_config (MCPAuthConfig object, see schema below)"
}

MCP Authentication Configuration

{
  type: "auth_type (enum: 'none' | 'api_key' | 'bearer' | 'basic' | 'custom_headers' | 'query_params')",
  api_key: "api_key_value (base64_string, optional, required when type='api_key')",
  header_name: "custom_header_name (string, optional, default: 'X-API-Key' for api_key type)",
  token: "bearer_token (jwt_string or access_token, optional, required when type='bearer')",
  username: "basic_auth_username (string, optional, required when type='basic')",
  password: "basic_auth_password (string, optional, required when type='basic')",
  headers: "custom_headers (key_value_object, optional, e.g., {'X-Custom': 'value', 'X-Version': '1.0'})",
  params: "query_parameters (key_value_object, optional, e.g., {'key': 'value', 'version': '2'})"
}

MCP Tools Configuration

{
  mcp_servers: "enabled_server_names (array of server_identifiers, optional, e.g., ['github_api', 'database_tools'])",
  builtin_tools: "included_builtin_tools (array of tool_names, optional, e.g., ['file_read', 'web_search'])",
  available_tools: "specific_tool_whitelist (array of tool_names, optional, overrides auto_discover)",
  auto_discover: "enable_auto_discovery (boolean, default: true, automatically includes all discovered tools)",
  filter: "excluded_tool_names (array of tool_names, optional, blacklist for auto-discovered tools)",
  tool_selection: "selection_mode (enum: 'auto' | 'manual', default: 'auto', how LLM selects tools)"
}

Core Components

Component Overview

Component	Purpose	Scope	Characteristics
Flow	Complete AI workflow definition	Global	Self-contained, executable, versioned
Node	Single processing step	Flow-scoped	Unique ID, input/output schema, configurable
Variable	Data binding mechanism	Flow/Node-scoped	Typed, resolvable, interpolated
Provider	External service integration	System-wide	Authenticated, rate-limited, standardized
Execution Context	Runtime state container	Execution-scoped	Isolated, temporary, tracked

Flow Structure

Element	Description	Required	Examples
Metadata	Name, version, description, author	✅	`"document_analyzer"`, `"v1.2.0"`
Variables	Input/output parameters with types	✅	`{"id": "user_input", "type": "string"}`
Nodes	Ordered processing steps	✅	`[{"id": "llm_1", "type": "LLM"}]`
Dependencies	Implicit from variable references	Auto-derived	`{{llm_1.output}} → dependency`

Variable Types

Variable Type	Purpose	Scope	Resolution
Flow Variables	Declared parameters with defaults	Flow-level	At flow definition
Node Outputs	Generated by node execution	Flow-level	Runtime, sequential
Input Variables	Runtime values from execution	Flow-level	At execution start
Scoped Variables	Temporary within control flow	Node-level	Within control structures

Provider Categories

Provider Type	Services	Capabilities	Integration Pattern
LLM Providers	OpenAI, Anthropic, Grok, Bedrock	Text generation, chat, vision	API key + endpoint
Vector Database	Pinecone, Weaviate, Chroma	Embedding CRUD, similarity search	Connection + index
Embedding	OpenAI, Nvidia, Microsoft	Text-to-vector conversion	Model selection + API
Custom	User-defined APIs	Domain-specific operations	Custom interface implementation

Execution Model

Concurrency Control

Aspect	Implementation	Configuration	Behavior
Global Limit	Max concurrent flows system-wide	`global_limit: 3`	Queue overflow flows
Flow Queue	In-memory FIFO queue	Auto-managed	Wait for available slots
Flow Isolation	Separate variable registries	Per-execution context	Prevent cross-flow contamination
Flow Identity	Timestamp + execution number	`20240115_143022_001`	Unique tracking identifier

Node Execution Patterns

Pattern	Behavior	Failure Handling	Use Cases
Sequential	Nodes execute in dependency order	Stop on first failure	Linear processing pipelines
Parallel Branches	Independent execution paths	Continue other branches	Concurrent operations
FOR_EACH	Sequential iteration processing	Stop on iteration failure	Batch processing
Conditional	Branch based on conditions	Skip false branches	Decision trees

Variable Resolution

Aspect	Mechanism	Format	Timing
Resolution	Just-in-time evaluation	`{{variable_name}}`	Before node execution
Registry	Flow-scoped storage	`{{node_id.output.property}}`	Runtime maintenance
Interpolation	Template string replacement	`{{llm_1.output.text}}`	Dynamic substitution
Isolation	Per-execution context	Separate namespaces	Prevent conflicts

Error and Resource Management

Management Type	Strategy	Scope	Cleanup
Error Handling	Fail-fast, no automatic retry	Node/Flow level	Immediate termination
Temporary Files	Auto-cleanup after completion	Flow execution	Post-execution cleanup
Registry State	Clear after completion	Per-execution	Memory management
Status Tracking	Real-time execution monitoring	System-wide	Observability support

Flow Definition Schema

Schema Overview

Every OpenFlow flow MUST conform to the following JSON schema structure. The schema enforces type safety, validates relationships between components, and ensures protocol compliance.

Root Flow Object

{
  name: "flow_name (alphanumeric_underscore, unique identifier, e.g., 'document_analysis_pipeline')",
  version: "semantic_version (semver format, e.g., '1.2.0', '2.0.0-beta.1')",
  description: "flow_description (detailed explanation of flow purpose and functionality)",
  author: "author_identifier (string, e.g., 'team@company.com' or 'john.doe')",
  variables: "flow_variables (array of FlowVariable objects, defines flow interface)",
  input: "input_variable_names (array of variable_ids that must be provided at runtime)",
  output: "output_variable_names (array of variable_ids that will be returned after execution)",
  nodes: "processing_nodes (array of FlowNode objects, ordered by dependencies)",
  metadata: "additional_metadata (key_value_object, optional, e.g., {category: 'nlp', complexity: 'medium'})",
  tags: "classification_tags (array of strings, optional, e.g., ['nlp', 'document-processing', 'ai'])",
  timeout: "execution_timeout_ms (integer, 10000-3600000, optional, max flow execution time)",
  concurrency: "concurrency_settings (ConcurrencyConfig object, optional, controls execution limits)"
}

FlowVariable Schema

Flow variables define the data interface of the flow:

{
  id: "variable_identifier (alphanumeric_underscore, unique within flow scope, e.g., 'input_document', 'llm_response')",
  type: "data_type (enum: 'string'|'number'|'boolean'|'array'|'object'|'file'|'null'|'any', default: 'string')",
  description: "variable_description (explains purpose and expected content, optional but recommended)",
  default: "default_value (any type matching the specified type annotation, optional)",
  required: "is_required (boolean, default: true if no default value provided, false if default exists)",
  validation: "validation_rules (ValidationRule object, optional, adds constraints and format requirements)"
}

Supported Data Types

Type	Description	Examples	Validation
`string`	Text values, template strings	`"Hello World"`, `"{{user_input}}"`	Length, pattern, format
`number`	Numeric values (int/float)	`42`, `3.14`, `-100`	Min/max, range validation
`boolean`	True/false values	`true`, `false`	Exact match only
`array`	Ordered collections	`["a", "b", "c"]`, `[1, 2, 3]`	Item type, min/max length
`object`	Key-value structures	`{"key": "value", "nested": {}}`	Property validation
`file`	File paths for processing	`"/path/to/file.pdf"`	Path validation, existence
`null`	Explicit null values	`null`	Null check only
`any`	Dynamic typing (use sparingly)	Any valid JSON	Minimal validation

Validation Rules

{
  pattern: "regex_pattern (string, for string validation, e.g., '^[a-zA-Z0-9_]+$', '^\\d{4}-\\d{2}-\\d{2}$')",
  minLength: "minimum_string_length (integer, 0-10000, for string types)",
  maxLength: "maximum_string_length (integer, 1-100000, for string types)",
  minimum: "minimum_numeric_value (number, for number types, e.g., 0, -100.5)",
  maximum: "maximum_numeric_value (number, for number types, e.g., 1000, 99.99)",
  enum: "allowed_values (array of valid options, e.g., ['draft', 'published', 'archived'])",
  format: "format_constraint (predefined format, e.g., 'email', 'url', 'date', 'uuid', 'base64')"
}

Core Concepts

On this page