Python Code Generation

Generate production-ready Python code from your visual workflows with asyncio support and comprehensive helper libraries.

Installation

Requirements

• Python 3.9+ (standard library only — no pip packages needed)
• Node.js installed and on PATH for Code nodes (execute_javascript)

# No dependencies to install — the helper library uses only the
# Python standard library. requirements.txt is included for tooling
# compatibility and contains no packages.
pip install -r requirements.txt

Environment Variables

export OPENAI_API_KEY=your_openai_api_key_here
# Or, when using Claude models:
export ANTHROPIC_API_KEY=your_anthropic_api_key_here

Usage Example

import asyncio
from circuitry import http_request, call_ai, execute_javascript, replace_template_variables
from circuitry_ui import show_message, confirm_action

# Generated function (example)
async def my_workflow(input_data: dict = {}) -> dict:
    data = input_data.copy()

    # HTTP request
    data = await http_request('https://api.example.com/data', method='GET')

    # User confirmation
    confirmation = await confirm_action(f"Process {len(data.get('items', []))} items?")
    if not confirmation['confirmed']:
        return {'cancelled': True}

    # AI processing (returns the response text as a string)
    prompt = replace_template_variables('Analyze this data: {{input}}', data)
    data['analysis'] = await call_ai(prompt, model='gpt-4o-mini')

    # Code node - execute JavaScript
    js_code = """
    data.processed = data.items.map(item => ({
        ...item,
        processed: true,
        timestamp: new Date().toISOString(),
        score: Math.random() * 100
    }));
    """
    data = await execute_javascript(js_code, data)

    # Show completion message
    await show_message(f"Processed {len(data['processed'])} items successfully", 'success')

    return data

# Use in your application
if __name__ == '__main__':
    result = asyncio.run(my_workflow({'items': [...]}))
    print(result)

Code Nodes

Python uses execute_javascript() for Code nodes. Node.js must be installed and on PATH:

# JavaScript code in Code node
js_code = """
data.processed = data.items.map(item => ({
    ...item,
    processed: true,
    timestamp: new Date().toISOString()
}));
"""

# Execute JavaScript with Python data
result = await execute_javascript(js_code, data)

Parallel Execution

Fork/join nodes use asyncio.gather():

import asyncio

# Fork node - parallel execution
async def parallel_branches(data):
    # Branch 1: API call
    async def branch1():
        return await http_request('https://api1.example.com/data')
    
    # Branch 2: AI processing (returns a string)
    async def branch2():
        return await call_ai('Process data', model='gpt-4o-mini')

    # Execute in parallel
    result1, result2 = await asyncio.gather(branch1(), branch2())

    # Join results
    return {
        **result1,
        'analysis': result2
    }

Helper Functions

HTTP Requests

# GET request — JSON responses are parsed automatically
data = await http_request('https://api.example.com/users')

# POST with data
result = await http_request(
    'https://api.example.com/users',
    method='POST',
    body={'name': 'John', 'email': 'john@example.com'},
    headers={'Authorization': 'Bearer token'}
)

AI Calls

call_ai() returns the response text as a string. Models containing "claude" route to Anthropic; everything else uses OpenAI-style chat completions (set endpoint for a local or self-hosted model).

# Simple AI call
response = await call_ai('Summarize this text', model='gpt-4o-mini')

# Claude model with a system prompt
analysis = await call_ai(
    'Analyze sentiment',
    model='claude-opus-4-8',
    system_prompt='You are a precise sentiment analyst.',
    max_tokens=500
)

# Local or self-hosted model (OpenAI-compatible endpoint)
local = await call_ai(
    'Classify this ticket',
    model='my-local-model',
    endpoint='http://localhost:11434/v1/chat/completions'
)

Template Variables

# {{input}} / {{previousOutput}} expand to the whole data object (JSON);
# dotted paths like {{input.user.name}} or {{user.name}} resolve inside it.
prompt = replace_template_variables('Hello {{user.name}}, data: {{input}}', data)

Deployment Options

FastAPI Server

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class WorkflowInput(BaseModel):
    data: dict

@app.post("/workflow")
async def run_workflow(input_data: WorkflowInput):
    try:
        result = await my_workflow(input_data.data)
        return {"success": True, "result": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

Docker Container

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt ./
RUN pip install -r requirements.txt
COPY workflow.py circuitry.py ./
CMD ["python", "workflow.py"]