Execution Nodes

nodes:
  - id: calculate_total
    type: code
    code:
      type: fixed
      value: |
        # Access state variables
        numbers = alita_state.get('numbers', [])
        
        # Calculate total
        total = sum(numbers)
        
        # Return result
        {"total": total}
    input:
      - numbers
    output:
      - total
    structured_output: true
    transition: END
state:
  input:
    type: str
  messages:
    type: list
  numbers:
    type: list
    value: []
  total:
    type: int
    value: 0

nodes:
  - id: process_user_data
    type: code
    code:
      type: fixed
      value: |
        import json
        
        # Access input data from state
        raw_data = alita_state.get('raw_data', [])
        min_score = alita_state.get('min_score', 50)
        
        # Process with error handling
        try:
            processed = []
            for item in raw_data:
                score = item.get('score', 0)
                if score >= min_score:
                    processed.append({
                        'id': item['id'],
                        'name': item['name'],
                        'grade': 'Pass',
                        'score': score
                    })
            
            {
                "processed_data": processed,
                "count": len(processed),
                "status": "success"
            }
        except Exception as e:
            {
                "error": str(e),
                "status": "failed",
                "count": 0
            }
    input:
      - raw_data
      - min_score
    output:
      - processed_data
      - count
      - status
    structured_output: true
    transition: END
state:
  input:
    type: str
  messages:
    type: list
  raw_data:
    type: list
    value: []
  min_score:
    type: int
    value: 50
  processed_data:
    type: list
    value: []
  count:
    type: int
    value: 0
  status:
    type: str
    value: ''

nodes:
  - id: fetch_api_data
    type: code
    code:
      type: fixed
      value: |
        import httpx
        import json
        
        query = alita_state.get('search_query', '')
        api_url = f"https://api.example.com/search?q={query}"
        
        try:
            # Use httpx for async HTTP calls (Pyodide compatible)
            async with httpx.AsyncClient() as client:
                response = await client.get(api_url, timeout=10.0)
                data = response.json()
            
            {
                "api_results": data.get('results', []),
                "result_count": len(data.get('results', [])),
                "status": "success"
            }
        except Exception as e:
            {
                "error": str(e),
                "status": "failed",
                "result_count": 0
            }
    input:
      - search_query
    output:
      - api_results
      - result_count
      - status
    structured_output: true
    transition: END
state:
  input:
    type: str
  messages:
    type: list
  search_query:
    type: str
    value: ''
  api_results:
    type: list
    value: []
  result_count:
    type: int
    value: 0
  status:
    type: str
    value: ''

nodes:
  - id: data_analysis
    type: code
    code:
      type: fixed
      value: |
        # Install numpy via micropip
        import micropip
        await micropip.install('numpy')
        import numpy as np
        
        # Access data from state
        values = alita_state.get('values', [])
        
        if not values:
            {"error": "No values provided", "status": "failed"}
        else:
            # Perform analysis with numpy
            data = np.array(values)
            
            {
                "mean": float(np.mean(data)),
                "median": float(np.median(data)),
                "std_dev": float(np.std(data)),
                "min": float(np.min(data)),
                "max": float(np.max(data)),
                "status": "success"
            }
    input:
      - values
    output:
      - mean
      - median
      - std_dev
      - min
      - max
      - status
    structured_output: true
    transition: END
state:
  input:
    type: str
  messages:
    type: list
  values:
    type: list
    value: []
  mean:
    type: float
    value: 0.0
  median:
    type: float
    value: 0.0
  std_dev:
    type: float
    value: 0.0
  min:
    type: float
    value: 0.0
  max:
    type: float
    value: 0.0
  status:
    type: str
    value: ''

nodes:
  - id: apply_business_rules
    type: code
    code:
      type: fixed
      value: |
        user_age = alita_state.get('user_age', 0)
        user_type = alita_state.get('user_type', 'guest')
        
        # Apply business logic
        if user_age < 18:
            category = "minor"
            allowed_actions = ["read", "comment"]
        elif user_type == "premium":
            category = "premium_adult"
            allowed_actions = ["read", "comment", "post", "moderate"]
        else:
            category = "standard_adult"
            allowed_actions = ["read", "comment", "post"]
        
        {
            "user_category": category,
            "allowed_actions": allowed_actions,
            "access_level": len(allowed_actions)
        }
    input:
      - user_age
      - user_type
    output:
      - user_category
      - allowed_actions
      - access_level
    structured_output: true
    transition: END
state:
  input:
    type: str
  messages:
    type: list
  user_age:
    type: int
    value: 0
  user_type:
    type: str
    value: 'guest'
  user_category:
    type: str
    value: ''
  allowed_actions:
    type: list
    value: []
  access_level:
    type: int
    value: 0

nodes:
  - id: list_artifacts
    type: code
    code:
      type: fixed
      value: |
        # Access Alita client (automatically injected when available)
        try:
            # List artifacts from a bucket
            bucket_name = alita_state.get('bucket_name', 'test')
            artifacts = alita_client.list_artifacts(bucket_name=bucket_name)
            
            {
                "artifacts": artifacts,
                "count": len(artifacts),
                "status": "success"
            }
        except Exception as e:
            {
                "error": str(e),
                "status": "failed",
                "count": 0
            }
    input:
      - bucket_name
    output:
      - artifacts
      - count
      - status
    structured_output: true
    transition: END
state:
  input:
    type: str
  messages:
    type: list
  bucket_name:
    type: str
    value: 'test'
  artifacts:
    type: list
    value: []
  count:
    type: int
    value: 0
  status:
    type: str
    value: ''

# Windows (PowerShell)
irm https://deno.land/install.ps1 | iex

# Bad - not serializable
class MyClass:
    pass
return {"result": MyClass()}

# Good - serializable
return {"result": {"name": "value", "count": 42}}

import micropip
await micropip.install('numpy')
import numpy as np

output:
  - result
  - status
structured_output: true

return {"result": 42, "status": "success"}

import httpx

async with httpx.AsyncClient() as client:
    response = await client.get(url)
    data = response.json()