Agent Flows Python Package

A powerful Python library for executing Agent Flows - visual, no-code workflows that integrate seamlessly into Python applications.

Features

• Multiple Executors: Built-in support for 14 different executor types (API calls, LLM instructions, MCP servers, etc.)
• Flow Initialization: Use the flow_variables step type (start remains as a deprecated alias) to seed runtime data consistently across flows.
• MCP Integration: Native support for Model Context Protocol (MCP) servers and tools
• Flexible Usage: Run flows from RealTimeX instances or local JSON files
• Simple API: Easy-to-use methods with explicit parameters
• CLI Interface: Command-line tools with uvx support (no installation required)
• Async & Sync: Both async/await and synchronous execution options
• Type Safety: Full type hints and Pydantic validation
• JSON Configuration: Simple, consistent configuration format
• Testing Framework: Deep pinning support for testing flows without API costs, including composite nodes (loops, conditionals)

Quick Start

Installation

# Install via pip
pip install agent-flows

# Or use with uvx for one-time execution (no installation required)
uvx agent-flows execute --flow-id <uuid> --api-key <key> --base-url <url>

Configuration

Configure using environment variables:

export AGENT_FLOWS_API_KEY="your-realtimex-api-key"
export AGENT_FLOWS_BASE_URL="https://your-realtimex-instance.com"
export LITELLM_API_KEY="your-llm-api-key"

# For MCP server integration
export MCP_ACI_API_KEY="your-aci-api-key"
export MCP_ACI_LINKED_ACCOUNT_OWNER_ID="your-linked-account-owner-id"

Usage

1. Execute Flows from RealTimeX

import asyncio
from agent_flows import FlowExecutor, AgentFlowsConfig, LiteLLMConfig

# Option 1: Use environment variables (recommended)
executor = FlowExecutor()

# Option 2: Create explicit configuration
system_config = AgentFlowsConfig(
    api_key="your-realtimex-api-key",
    base_url="https://your-realtimex-instance.com",
    litellm=LiteLLMConfig(
        api_key="your-llm-api-key",
        api_base="https://api.openai.com/v1"
    )
)
executor = FlowExecutor(config=system_config)

# Execute flow with explicit parameters
result = await executor.execute_flow(
    flow_id="550e8400-e29b-41d4-a716-446655440000",
    variables={"input_param": "value", "max_results": 10}
)

print(f"Success: {result.success}")
print(f"Steps executed: {result.steps_executed}")
print(f"Execution time: {result.execution_time:.2f}s")

2. Execute Local Flow Files

from agent_flows import FlowExecutor

# Create executor from local flow file (uses environment variables for LLM/API config)
executor = FlowExecutor.from_file(
    flow_file="examples/1_basic_executors/start.json"
)

# Execute with explicit parameters
result = await executor.execute_flow(
    variables={"name": "World", "greeting": "Hello"}
)

print(f"Local flow result: {result.success}")

3. Synchronous Execution (No async/await)

from agent_flows import FlowExecutor

# Create executor (uses environment variables)
executor = FlowExecutor.from_file(flow_file="examples/simple_flow.json")

# Run synchronously with explicit parameters
result = executor.run(
    variables={"name": "World"}
)

print(f"Success: {result.success}")

4. Configuration Options

from agent_flows import FlowExecutor, AgentFlowsConfig

# Option 1: From environment variables (recommended)
executor = FlowExecutor()  # Uses AGENT_FLOWS_* env vars

# Option 2: Explicit configuration
system_config = AgentFlowsConfig(
    api_key="your-key",
    base_url="https://your-instance.com"
)
executor = FlowExecutor(config=system_config)

# Option 3: From flow dictionary
flow_dict = {"uuid": "123", "name": "Test", "steps": [...]}
executor = FlowExecutor.from_dict(flow_dict=flow_dict)

CLI Usage

Execute flows from RealTimeX

# Execute with environment variables (recommended)
agent-flows execute \
  --flow-id "550e8400-e29b-41d4-a716-446655440000"

# Execute with inline config
agent-flows execute \
  --flow-id "550e8400-e29b-41d4-a716-446655440000" \
  --config '{"api_key":"key","base_url":"url","litellm":{"api_key":"llm-key"}}'

# Execute with variables
agent-flows execute \
  --flow-id "550e8400-e29b-41d4-a716-446655440000" \
  --variables '{"input_text":"Hello World","max_results":10}'

Execute local flow files

# Run a local flow file
agent-flows run examples/1_basic_executors/start.json

# Run with variables (JSON string)
agent-flows run examples/3_workflows/2_breaking_news_to_linkedin/flow.json \
  --variables '{"name":"John","age":30}'

# Run with variables from file
agent-flows run examples/3_workflows/2_breaking_news_to_linkedin/flow.json \
  --variables-file examples/3_workflows/2_breaking_news_to_linkedin/variables.json

# Run with explicit config for LLM settings
agent-flows run examples/3_workflows/2_breaking_news_to_linkedin/flow.json \
  --config '{"litellm":{"api_key":"your-llm-key"}}'

Using uvx (no installation required)

# Execute a flow with uvx
uvx agent-flows execute \
  --flow-id "550e8400-e29b-41d4-a716-446655440000" \
  --config '{"api_key":"key","base_url":"url","litellm":{"api_key":"llm-key"}}'

# Run a local flow with uvx
uvx agent-flows run examples/1_basic_executors/start.json \
  --variables '{"greeting":"Hello from uvx!"}'

Supported Flow Types

The package includes 14 built-in executors:

• Flow Variables: Variable initialization and flow setup
• API Call: HTTP requests with full REST API support and retry logic
• LLM Instruction: Multi-provider LLM integration (OpenAI, Anthropic, etc.)
• Web Scraping: Content extraction with CSS selectors and auto-summarization
• Conditional: Branching logic with multiple comparison operators
• Switch: Multi-case routing based on variable values
• Loop: Iterative execution with for/while/forEach support and clean output format
• MCP Server Action: Execute actions on Model Context Protocol (MCP) servers

For detailed documentation on each executor, see docs/executors/.

MCP Integration

Agent Flows provides native support for Model Context Protocol (MCP) servers, enabling seamless integration with external tools and services.

MCP Server Examples

# Send email via Gmail MCP server
{
  "type": "mcpServerAction",
  "config": {
    "provider": "remote",
    "serverId": "GMAIL",
    "action": "GMAIL__SEND_EMAIL",
    "params": {
      "to": "recipient@example.com",
      "subject": "Hello from Agent Flows",
      "body": "This email was sent via MCP integration!"
    },
    "resultVariable": "email_result"
  }
}

# Upload file to Google Drive
{
  "type": "mcpServerAction",
  "config": {
    "provider": "remote",
    "serverId": "GOOGLE_DRIVE",
    "action": "DRIVE__UPLOAD_FILE",
    "params": {
      "fileName": "report.pdf",
      "fileContent": "${file_data}",
      "folderId": "${target_folder}"
    },
    "resultVariable": "upload_result"
  }
}

Available MCP Servers

• Gmail: Email operations (send, read, manage)
• Google Drive: File operations (upload, download, organize)
• Google Calendar: Calendar management (create events, schedule meetings)
• Slack: Messaging and channel operations
• Database: Query execution and data management
• File Manager: Local file system operations

MCP Configuration

Set up MCP integration using environment variables or configuration files:

# Environment variables
export MCP_ACI_API_KEY="your-aci-api-key"
export MCP_ACI_LINKED_ACCOUNT_OWNER_ID="your-linked-account-owner-id"

Or in your configuration:

from agent_flows import AgentFlowsConfig
from agent_flows.models.config import MCPConfig

config = AgentFlowsConfig(
    mcp=MCPConfig(
        aci_api_key="your-aci-api-key",
        aci_linked_account_owner_id="your-linked-account-owner-id"
    )
)

Architecture

The package follows a modular, extensible architecture:

agent_flows/
├── core/           # Core execution engine
├── executors/      # Block-specific executors
├── models/         # Pydantic data models
├── api/           # API client for flow fetching
├── cli/           # Command-line interface
└── utils/         # Utility functions

API Reference

FlowExecutor Methods

# Constructor
FlowExecutor(
    config=None,           # AgentFlowsConfig: System configuration (uses env vars if None)
    flow_config=None,      # FlowConfig: Local flow configuration
    log_level="INFO",      # str: Logging level (e.g., "DEBUG", "INFO", "WARNING")
    log_json_format=False  # bool: Format logs as JSON
)

# Class methods
FlowExecutor.from_file(
    flow_file,            # str: Path to flow JSON file
    config=None,          # AgentFlowsConfig: System configuration (uses env vars if None)
)

FlowExecutor.from_dict(
    flow_dict,            # dict: Flow configuration dictionary
    config=None,          # AgentFlowsConfig: System configuration (uses env vars if None)
)

# Execution methods
await executor.execute_flow(
    flow_id=None,         # str: Flow UUID (optional if flow_config provided)
    variables=None,       # dict: Flow variables
    context=None          # dict: Additional execution context
)

executor.run(
    flow_id=None,         # str: Flow UUID (optional if flow_config provided)
    variables=None        # dict: Flow variables
)

Configuration Types

System Configuration (AgentFlowsConfig): API credentials and LLM settings Flow Configuration (FlowConfig): The actual flow definition with steps

Demo

Try the comprehensive demo:

python demo_packaging.py

Requirements

• Python 3.8+
• Dependencies: aiohttp, pydantic, click, litellm

License

MIT License