Code Blocks¶

A code block component is an executable unit of code managed by the platform. Code blocks support multiple languages, have built-in syntax validation, Git-backed version history, and can be referenced by tools and workflows.

Supported Languages¶

Language	Extension	Syntax Validation	Runtime
Python	`.py`	Yes	Managed container
TypeScript	`.ts`	Planned	Managed container
JavaScript	`.js`	Planned	Managed container
SQL	`.sql`	No	Database connector
Bash	`.sh`	No	Managed container

Creating a Code Block¶

PythonUI

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

code_block = client.code_blocks.create({
    "name": "Clean Customer Data",
    "slug": "clean-customer-data",
    "description": "Normalizes and deduplicates customer records",
    "language": "python",
    "code": """
def main(records: list[dict]) -> list[dict]:
    \"\"\"Clean and deduplicate customer records.\"\"\"
    seen = set()
    cleaned = []
    for record in records:
        email = record.get("email", "").strip().lower()
        if email and email not in seen:
            seen.add(email)
            cleaned.append({
                "email": email,
                "name": record.get("name", "").strip().title(),
                "phone": normalize_phone(record.get("phone", "")),
            })
    return cleaned


def normalize_phone(phone: str) -> str:
    digits = "".join(c for c in phone if c.isdigit())
    if len(digits) == 10:
        return f"+1{digits}"
    return f"+{digits}" if digits else ""
""",
    "entry_point": "main",
    "dependencies": ["pydantic>=2.0"],
    "tags": ["data-cleaning", "etl"],
})

Navigate to Components > Code Blocks > New. Select language, enter code, set entry point. Click Create.

Configuration¶

Field	Type	Required	Description
`name`	string	Yes	Display name
`slug`	string	Yes	URL-safe identifier
`language`	enum	Yes	`python`, `typescript`, `javascript`, `sql`, `bash`
`code`	string	Yes	Source code content
`entry_point`	string	No	Main function or method name (e.g., `main`)
`dependencies`	string[]	No	Package dependencies (e.g., `["requests>=2.28", "pandas"]`)
`description`	string	No	What this code block does
`tags`	string[]	No	Tags for filtering

Input and Output¶

Code blocks communicate through their entry point function's parameters and return value. The platform calls the entry_point function with the execution input data and captures the return value as output.

def main(customer_id: str, include_history: bool = False) -> dict:
    """Fetch customer profile.

    Args:
        customer_id: The customer's unique identifier.
        include_history: Whether to include purchase history.

    Returns:
        Customer profile data.
    """
    profile = fetch_profile(customer_id)
    if include_history:
        profile["history"] = fetch_history(customer_id)
    return profile

When a code block is used as a Tool, the tool's parameter definitions map to the function's arguments. When used in a Workflow, the node's input data is passed as keyword arguments.

Syntax Validation¶

For Python code blocks, the platform validates syntax before saving. Validation checks for:

Python syntax errors (parse errors)
Entry point existence (the specified function must be defined in the code)
Basic security patterns

Coming Soon

The Python SDK for local development is not yet publicly available.

# Validate code without saving
from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

code = """
def main(data):
    return data.upper(
"""

validation = client.code_blocks.validate_code(component_id, code, "python")

print(f"Valid: {validation.valid}")
print(f"Errors: {validation.errors}")
# Valid: False
# Errors: ["SyntaxError: unexpected EOF while parsing (line 3)"]

Validation scope

Syntax validation currently only checks Python code. TypeScript, JavaScript, SQL, and Bash validation is planned for a future release. The code is still saved for these languages but without syntax checks.

Code Editor and Git Storage¶

Code blocks are stored in a Git repository scoped to your organization and workspace. Every save creates a Git commit, giving you a complete audit trail.

Saving Code¶

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

result = client.code_blocks.save_code(component_id, updated_code, "python")

print(f"Commit: {result.commit_hash}")

Viewing History¶

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

history = client.code_blocks.get_code_history(component_id)

for entry in history.entries:
    print(f"{entry.commit_hash[:8]} — {entry.message} ({entry.author})")

Reverting to a Previous Version¶

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

client.code_blocks.revert_code(component_id, "abc12345")

Runtime Environment¶

When a code block is deployed, the platform creates a managed container with:

The specified language runtime
All declared dependencies installed
Network access to platform services (connectors, datasets, other components)
Resource limits based on the deployment ring configuration

Dependencies¶

Declare package dependencies as pip-style requirement strings:

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

code_block = client.code_blocks.create({
    "name": "Data Analyzer",
    "slug": "data-analyzer",
    "language": "python",
    "code": "...",
    "dependencies": [
        "pandas>=2.0,<3.0",
        "numpy>=1.24",
        "scikit-learn>=1.3",
    ],
})

Execution¶

Direct Execution¶

Execute a code block directly for testing:

from flow_sdk import FlowSDK

sdk = FlowSDK()

result = await sdk.code_blocks.run(
    code_block_id=code_block.id,
    input_parameters={
        "records": [
            {"email": "Alice@Example.com", "name": "alice smith", "phone": "5551234567"},
            {"email": "bob@example.com", "name": "BOB JONES", "phone": "(555) 987-6543"},
            {"email": "alice@example.com", "name": "Alice Smith", "phone": "555-123-4567"},
        ]
    },
)

print(f"Status: {result.status}")
print(f"Output: {result.output_data}")
# Output: [
#   {"email": "alice@example.com", "name": "Alice Smith", "phone": "+15551234567"},
#   {"email": "bob@example.com", "name": "Bob Jones", "phone": "+15559876543"}
# ]

As a Workflow Node¶

Code blocks can be referenced as nodes in workflows:

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

workflow = client.workflows.create({
    "name": "ETL Pipeline",
    "slug": "etl-pipeline",
    "trigger_type": "schedule",
    "trigger_config": {"cron_expression": "0 2 * * *"},
    "nodes": [
        {"id": "extract", "type": "codeblock",
         "component_ref": extract_block.id, "label": "Extract Data"},
        {"id": "transform", "type": "codeblock",
         "component_ref": transform_block.id, "label": "Transform"},
        {"id": "load", "type": "codeblock",
         "component_ref": load_block.id, "label": "Load to Warehouse"},
    ],
    "edges": [
        {"source": "extract", "target": "transform"},
        {"source": "transform", "target": "load"},
    ],
})

Usage Tracking¶

Track which components reference a code block:

Coming Soon

The Python SDK for local development is not yet publicly available.

from flow_sdk.cli_client import CLIClient

client = CLIClient(config)

used_by = client.code_blocks.get_used_by(component_id)

print(f"Used by {used_by.total_references} components:")
for tool in used_by.tools:
    print(f"  Tool: {tool.name}")