Code Executor Tools

​

Overview

​

Tool Name

code_executor_tools

​

Purpose

​

Functions Available

1. execute_python_code: Run Python locally or on Snowflake Snowpark. For local execution, assign the final value to the result variable.
2. create_env: Create a named managed environment for local or Snowpark execution.
3. list_envs: List available managed environments.
4. list_env_deps: Inspect the packages installed in one managed environment.
5. update_env: Add or remove packages from an existing managed environment.
6. delete_env: Delete a managed environment after explicit confirmation.
7. run_repo_script: Run a server-pinned Python required_check for the active Blueprint phase and record platform-owned evidence.

​

Key Features

​

Managed local environments

​

run_repo_script

run_repo_script(check_name="validate_repository")

create_env(
    name="synth-gen",
    deps=["faker>=20.0.0", "pandas>=2.0.0", "numpy>=1.24.0"],
)

list_env_deps(venv="synth-gen")

execute_python_code(
    code="import faker, pandas, numpy; result = 'ok'",
    env="synth-gen",
)

update_env(
    name="synth-gen",
    add_packages=["pyarrow>=14.0.0"],
)

​

Genesis SDK in local execution

from genesis_sdk import Genesis

gx = Genesis.from_environment()
gx.progress.message("Starting query")

rows = gx.query.rows(
    connection_id="snowflake.prod",
    sql="select region, sum(revenue) as revenue from sales group by region",
    max_rows=1000,
)

gx.artifacts.write_json("outputs/regions.json", rows)
gx.progress.complete("query")
result = {"row_count": len(rows), "artifact": "outputs/regions.json"}

• gx.capabilities()
• gx.query.rows(...) and gx.query.df(...)
• gx.tools.list(), gx.tools.schema(name), and gx.tools.call(name, args)
• gx.artifacts.write_text(...), write_json(...), write_csv(...), write_bytes(...), read_text(...), read_json(...), read_bytes(...), list(...), exists(...), and delete(...)
• gx.progress.message(...), update(...), complete(...), fail(...), and artifact(...)

​

Workspace files (local execution)

from genesis import BOT_GIT_PATH, BOT_REPOS_PATH, BOT_SCRATCH_PATH, sanitize_path

path = f"{BOT_REPOS_PATH}/default/report.csv"   # repo working tree
with open(path, "w") as f:
    f.write(csv_text)
result = {"path": sanitize_path(path)}          # -> "BOT_REPOS_PATH/default/report.csv"

​

Input Parameters for Each Function

​

python_exec

​

Use Cases

1. Ad-Hoc Data Analysis Perform quick calculations, statistical analysis, or data exploration without creating permanent functions or scripts.
2. Data Transformation Execute custom transformation logic on datasets that does not fit into standard SQL or built-in functions.
3. API Integration Make HTTP requests, parse responses, and integrate with external APIs dynamically based on user requirements.
4. Prototyping & Testing Test algorithms, validate assumptions, or prototype solutions before implementing them as permanent features.
5. Custom Business Logic Implement organization-specific calculations, rules, or validations that vary by use case or customer.

​

Workflow/How It Works

1. Step 1: Simple Calculation Execute basic Python for quick results:
   python_exec(
       code="result = 42 * 365; print(f'Days in 42 years: {result}')"
   )
   
2. Step 2: Data Analysis with Pandas Analyze data using pre-loaded libraries:
   python_exec(
       code="""
   import pandas as pd
   import numpy as np
   
   data = {
       'product': ['A', 'B', 'C', 'D', 'E'],
       'sales': [150, 200, 175, 300, 225],
       'cost': [100, 150, 125, 200, 175]
   }
   
   df = pd.DataFrame(data)
   df['profit'] = df['sales'] - df['cost']
   df['margin'] = (df['profit'] / df['sales'] * 100).round(2)
   
   summary = {
       'total_sales': df['sales'].sum(),
       'total_profit': df['profit'].sum(),
       'avg_margin': df['margin'].mean(),
       'top_product': df.loc[df['profit'].idxmax(), 'product']
   }
   
   print("Sales Analysis:")
   print(df)
   print(f"Summary: {summary}")
   """
   )
   
3. Step 3: Pass Variables from Context Use data from previous operations:
   python_exec(
       code="""
   import pandas as pd
   
   df = pd.DataFrame(customer_data)
   
   df['segment'] = pd.cut(
       df['total_spend'], 
       bins=[0, 1000, 5000, float('inf')],
       labels=['Bronze', 'Silver', 'Gold']
   )
   
   segment_summary = df.groupby('segment').agg({
       'customer_id': 'count',
       'total_spend': 'sum'
   }).round(2)
   
   print(segment_summary)
   result = segment_summary.to_dict()
   """,
       variables={'customer_data': customer_data},
       return_var='result'
   )
   
4. Step 4: API Integration Make external API calls and process responses:
   python_exec(
       code="""
   import os
   import requests
   
   api_key = os.environ['SECRET_OPENWEATHER_API_KEY']
   
   response = requests.get(
       'https://api.openweathermap.org/data/2.5/weather',
       params={
           'q': city_name,
           'appid': api_key,
           'units': 'metric'
       }
   )
   
   if response.status_code == 200:
       weather_data = response.json()
       result = {
           'temperature': weather_data['main']['temp'],
           'humidity': weather_data['main']['humidity'],
           'description': weather_data['weather'][0]['description']
       }
       print(f"Weather in {city_name}: {result['temperature']}°C")
   else:
       result = {'error': f"API call failed: {response.status_code}"}
   """,
       variables={'city_name': 'London'},
       secrets=['secret_openweather_api_key'],
       return_var='result'
   )
   
5. Step 5: File Operations Read and write files within execution:
   python_exec(
       code="""
   import json
   from genesis import BOT_REPOS_PATH, sanitize_path
   
   config_path = f"{BOT_REPOS_PATH}/default/config.json"
   with open(config_path, 'r') as f:
       config = json.load(f)
   
   config['last_run'] = '2024-01-15'
   config['status'] = 'completed'
   
   with open(config_path, 'w') as f:
       json.dump(config, f, indent=2)
   
   print(f"Config updated: {config}")
   result = {'config': config, 'path': sanitize_path(config_path)}
   """,
       return_var='result'
   )
   
6. Step 6: Complex Data Processing Implement multi-step algorithms:
   python_exec(
       code="""
   import pandas as pd
   import numpy as np
   
   df = pd.DataFrame(transaction_data)
   df['date'] = pd.to_datetime(df['date'])
   df = df.sort_values('date')
   
   df['rolling_avg_7d'] = df['amount'].rolling(window=7).mean()
   df['rolling_sum_30d'] = df['amount'].rolling(window=30).sum()
   
   mean = df['amount'].mean()
   std = df['amount'].std()
   df['z_score'] = (df['amount'] - mean) / std
   df['is_anomaly'] = abs(df['z_score']) > 3
   
   anomalies = df[df['is_anomaly']]
   summary = {
       'total_transactions': len(df),
       'anomaly_count': len(anomalies),
       'anomaly_dates': anomalies['date'].dt.strftime('%Y-%m-%d').tolist(),
       'avg_7d_trend': df['rolling_avg_7d'].iloc[-1]
   }
   
   print(f"Detected {len(anomalies)} anomalies")
   result = summary
   """,
       variables={'transaction_data': transactions},
       return_var='result',
       timeout=60
   )
   

​

Integration Relevance

• data_connector_tools to fetch data from databases, then process with custom Python code.
• file tools to read/write files before or after code execution.
• github_connector_tools / git_and_file_tools (git_action, for GitLab) to version control generated scripts.
• image_tools to process images or generate visualizations from code execution results.
• web_access_tools to fetch data via HTTP within executed code.
• project_manager_tools to execute custom code as part of automated mission tasks.

​

Configuration Details

• Execution Environment: Python 3.9+ with isolated namespace per execution.
• Pre-loaded Libraries: pandas, numpy, requests, json, datetime, os, sys, re, math, statistics.
• Resource Limits: Local execution time limit (default 10s, max 600s), address-space cap 1,073,741,824 bytes, max file size 52,428,800 bytes, max 128 open files, and max 8 processes.
• File System Access: Reads and writes are confined to BOT_GIT_PATH, BOT_REPOS_PATH, and BOT_SCRATCH_PATH (import via from genesis import ...). Local execution may read task-owned documents and data in the task’s own repository or scratch area, but not source-code files or sensitive files such as other threads’ uploads, credentials, secrets, or .git internals. Local runs start in BOT_SCRATCH_PATH, which is also the temp root. You may write into the repo working tree under BOT_REPOS_PATH as well as scratch. Genesis stages a managed child_runner.py in scratch for each local run. Absolute workspace paths in responses are rewritten to BOT_*_PATH shorthands, and interpreter-installation paths to PYTHON_ROOT, before they reach the agent.
• Network Access: HTTP/HTTPS requests allowed; other protocols restricted.
• Security Mode: Blocks process creation, native library / FFI loading, dangerous filesystem probes, and other common sandbox-escape patterns. This is defense in depth, not a kernel-enforced boundary.
• Return Value: Can return primitive types, dictionaries, lists, or serializable objects.

​

Limitations or Notes

1. Execution Timeout: Local execution defaults to 10 seconds and can run for up to 600 seconds (10 minutes). Long-running operations may be terminated once the timeout is reached.
2. Memory Limits: Default address-space cap is 1,073,741,824 bytes (~1 GB). Large dataset operations may exceed limits and fail.
3. No Subprocess: Cannot spawn subprocesses, fork, or execute shell commands from local execution.
4. No Native / FFI Loads: Native library loading and common FFI bridges such as ctypes and cffi are blocked in local execution.
5. Managed Dependencies: Do not install packages from inside the running code. For local execution, create or update a named environment first, then rerun with env="<name>".
6. Local Inline Entry Point: Local execution must use inline code; script_path is rejected, and local code cannot load or execute code from files. Use Snowpark for file-backed code execution.
7. Network Limitations: Only HTTP/HTTPS allowed; no raw socket access or other protocols.
8. File System Scope: Reads and writes are confined to BOT_GIT_PATH, BOT_REPOS_PATH, and BOT_SCRATCH_PATH (from from genesis import ...); task-owned documents and data may be read, but source-code and sensitive files are disallowed, and the sandbox refuses paths outside those roots. Low-level file descriptor access is blocked, and path-based operations such as os.rename, os.replace, os.link, os.symlink, os.readlink, os.access, os.chmod, os.truncate, and os.utime are enforced against the same allowed-path rules. Call sanitize_path(p) when returning file locations so output stays host-agnostic.
9. Serialization: Return values must be JSON-serializable or primitive types.
10. State Persistence: Each execution is isolated; no state persists between calls.
11. Large Responses: When the serialized execute_python_code response exceeds the inline Python budget (default 98,304 bytes / 96 KB, configurable with GENESIS_TOOL_OUTPUT_PYTHON_BUDGET), Genesis returns a capped envelope with tool_output_capped, preview, truncation_info, full_output_ref, and agent_guidance instead of the full inline payload.

​

Supported Operations

​

Not Supported

​

Output

• Success: Execution result with returned variable(s), stdout output, and execution time.
• Variables: Dictionary of all variables in execution context (unless specific return_var specified).
• Standard Output: Captured print statements and logging output. For local execution, stdout and stderr are each capped at 200,000 characters; larger streams are truncated with ... [truncated] before the normal inline response budget is applied.
• Capped Oversized Output: truncation_info includes bucket, bytes_in, bytes_out, budget_bytes, and spill_path. When full_output_ref is not null, the full output is saved under truncated_tool_outputs/<thread_id>/... in the apps BotGit repo and can be paged back with file(action="read", file_path=full_output_ref, repo_id="apps", start_byte=0, max_bytes=N), continuing from each response’s continuation.next_start_byte. The byte range is the read shape that returns the saved copy exactly as written; the whole-file and line-window paths read in text mode and rewrite line endings.
• The accompanying agent_guidance explains that oversized output is routinely stored this way, is not a context limit or an error, and can be read back from the spill path without data loss. If saving the spill copy fails, it instead says that no spill file is available and recommends rerunning with narrower parameters.
• Truncation record: Over-budget JSON objects are shrunk in place when possible: every key at every depth is preserved, and trailing elements are removed from the longest lists first. The returned object includes tool_output_truncation with truncated, truncation_basis, truncated_keys, bytes_before, bytes_after, full_output_ref, and agent_guidance. For each shortened list, truncated_keys reports its JSON key path as an array plus the kept and total element counts. This means a missing key was not silently dropped by the cap; only reported list elements may be absent. The governor measures candidate payloads against the configured byte budget and keeps as many list elements as fit; the result is not decided by a fixed count of restoration attempts. Use bytes_after and each entry’s kept/total counts to see the actual delivered size and what was omitted.
• Amended next step: When a shrunk object carries a next call shape (a verb plus a string why — the directive the tool computed over the whole result), the shortening is stated at the front of why: which lists were shortened with their kept/total counts, and the way to the rest (the full_output_ref path with the file(action="read", ...) call that reads it, or advice to narrow the call when no spill copy was saved). verb, args and the tool’s own sentence are preserved. A next that is not a call shape is left untouched.
• Tail-cut fallback: Non-object payloads and objects that cannot be shrunk under the budget keep the capped envelope (tool_output_capped, preview, truncation_info, full_output_ref, and agent_guidance) and also include the same tool_output_truncation record. Its truncation_basis identifies why the fallback was used (not_an_object, object_irreducible, record_key_in_use, serialized_key_collision, or structure_unwalkable), and a compact copy of the record’s measurements is appended on a trailer line in preview when that trailer fits within the budget.
• Execution Time: Time taken to execute code in seconds.
• Errors: Exception type, error message, and full stack trace for debugging.
• Local Failure Metadata: Failed local runs can also include status, error_type, policy_violations, and metrics. error_type is the reported exception class name. status is one of user_error, policy_denied, resource_limit, or internal_error. metrics includes wall_ms, cpu_ms, peak_rss_bytes, applied_limits, and skipped_limits.
• Safety-Judge Denials: A request denied by the local safety judge includes a fixed, actionable safety_guidance.reason instead of the judge’s free-form explanation. judge_reason_code identifies the category: OUTSIDE_ROOTS, SOURCE_READ, CODE_EXEC_FROM_FILE, PROCESS_OR_ESCAPE, NATIVE_BYPASS, HOST_PROBE, NETWORK, SENSITIVE_FILE, or OTHER. NETWORK indicates code tried to open a socket, HTTP, or database connection to a private, internal, or Outpost-routed host; use a Genesis connection to reach that system instead.
• Safety-Judge Unavailable: If the local safety judge fails or returns no decision, safety_guidance identifies this as an outage. For a request-side failure, the code does not run; for an output-side failure, the code has already run and its output is withheld, so do not retry it because its side effects may have happened. Operators can check the active provider’s model assignments and the server log’s sanitized judge_error diagnostic.
• Path Sanitization: When a response contains absolute workspace paths, the local executor rewrites them to BOT_GIT_PATH / BOT_REPOS_PATH / BOT_SCRATCH_PATH shorthands and may set paths_sanitized: true with a notice suggesting sanitize_path(p) for future runs.
• AWS/EKS Sandbox Guidance: When the request or response mentions boto3, AWS SDK work, EKS/Kubernetes, Helm, kubectl, AWS CLI, cloud deployment, or cloud monitoring, Genesis can add agent_reminders, sandbox_tools_recommended, and sandbox_tools_reference to recommend sandbox_action plus the aws-eks-sandbox-operations skill.
• Warnings: Any resource warnings (approaching timeout, high memory usage).

​

Best Practices

​

Example: Complete Data Analysis Workflow

# Step 1: Fetch data from database
query_result = _query_database(
    connection_id="analytics_db",
    query="SELECT customer_id, order_date, order_amount, product_category FROM orders WHERE order_date >= '2024-01-01' LIMIT 10000",
    max_rows=10000
)

# Step 2: Perform complex analysis
analysis_result = python_exec(
    code="""
import pandas as pd
import numpy as np

df = pd.DataFrame(data)
df['order_date'] = pd.to_datetime(df['order_date'])
df['order_amount'] = pd.to_numeric(df['order_amount'], errors='coerce')
df = df.dropna()

print(f"Loaded {len(df)} orders")

customer_stats = df.groupby('customer_id').agg({
    'order_amount': ['sum', 'mean', 'count'],
    'order_date': ['min', 'max']
}).reset_index()

customer_stats.columns = ['customer_id', 'total_spend', 'avg_order', 'order_count', 'first_order', 'last_order']
customer_stats['lifetime_days'] = (customer_stats['last_order'] - customer_stats['first_order']).dt.days

def segment_customer(row):
    if row['total_spend'] > 5000:
        return 'VIP'
    elif row['total_spend'] > 2000:
        return 'Gold'
    elif row['total_spend'] > 500:
        return 'Silver'
    else:
        return 'Bronze'

customer_stats['segment'] = customer_stats.apply(segment_customer, axis=1)

category_stats = df.groupby('product_category').agg({
    'order_amount': ['sum', 'mean', 'count'],
    'customer_id': 'nunique'
}).reset_index()

category_stats.columns = ['category', 'total_revenue', 'avg_order', 'order_count', 'unique_customers']
category_stats = category_stats.sort_values('total_revenue', ascending=False)

daily_sales = df.groupby(df['order_date'].dt.date)['order_amount'].agg(['sum', 'count']).reset_index()
daily_sales.columns = ['date', 'revenue', 'orders']
daily_sales['avg_order_value'] = (daily_sales['revenue'] / daily_sales['orders']).round(2)
daily_sales['revenue_7d_avg'] = daily_sales['revenue'].rolling(window=7).mean().round(2)
daily_sales['revenue_growth'] = daily_sales['revenue'].pct_change() * 100

summary = {
    'total_revenue': float(df['order_amount'].sum()),
    'total_orders': len(df),
    'unique_customers': df['customer_id'].nunique(),
    'avg_order_value': float(df['order_amount'].mean()),
    'segments': customer_stats['segment'].value_counts().to_dict(),
    'top_category': category_stats.iloc[0]['category'],
    'top_category_revenue': float(category_stats.iloc[0]['total_revenue'])
}

result = {
    'summary': summary,
    'customer_segments': customer_stats.to_dict('records'),
    'category_performance': category_stats.to_dict('records'),
    'daily_trends': daily_sales.tail(30).to_dict('records')
}
""",
    variables={'data': query_result['data']},
    return_var='result',
    timeout=60
)

print(f"Summary: {analysis_result['result']['summary']}")

​

Advanced Features

​

Statistical Analysis

python_exec(
    code="""
import numpy as np
from scipy import stats

data = np.array(values)

result = {
    'mean': float(np.mean(data)),
    'median': float(np.median(data)),
    'std': float(np.std(data)),
    'min': float(np.min(data)),
    'max': float(np.max(data)),
    'q25': float(np.percentile(data, 25)),
    'q75': float(np.percentile(data, 75))
}

stat, p_value = stats.normaltest(data)
result['is_normal'] = p_value > 0.05
result['normality_p_value'] = float(p_value)

print(f"Distribution analysis: {result}")
""",
    variables={'values': [23, 45, 67, 89, 34, 56, 78, 90, 12, 45]},
    return_var='result'
)

​

Text Processing

python_exec(
    code="""
import re
from collections import Counter

text = text_data.lower()
text = re.sub(r'[^a-z0-9 ]', '', text)
words = text.split()

stop_words = {'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for'}
words = [w for w in words if w not in stop_words and len(w) > 2]

word_freq = Counter(words)
top_words = word_freq.most_common(20)

result = {
    'total_words': len(words),
    'unique_words': len(set(words)),
    'top_words': [{'word': w, 'count': c} for w, c in top_words]
}

print(f"Analyzed {result['total_words']} words")
""",
    variables={'text_data': long_text_content},
    return_var='result'
)

​

Data Validation

python_exec(
    code="""
import pandas as pd

df = pd.DataFrame(dataset)

validation_results = {
    'row_count': len(df),
    'column_count': len(df.columns),
    'missing_values': df.isnull().sum().to_dict(),
    'duplicate_rows': df.duplicated().sum(),
    'data_types': df.dtypes.astype(str).to_dict()
}

errors = []

required_cols = ['customer_id', 'order_date', 'amount']
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
    errors.append(f"Missing required columns: {missing_cols}")

if 'amount' in df.columns:
    negative_amounts = (df['amount'] < 0).sum()
    if negative_amounts > 0:
        errors.append(f"Found {negative_amounts} negative amounts")

if 'order_date' in df.columns:
    try:
        pd.to_datetime(df['order_date'])
    except:
        errors.append("Invalid date format in order_date column")

validation_results['errors'] = errors
validation_results['is_valid'] = len(errors) == 0

print(f"Validation {'passed' if validation_results['is_valid'] else 'failed'}")
result = validation_results
""",
    variables={'dataset': data_to_validate},
    return_var='result'
)

​

API Pagination Handler

python_exec(
    code="""
import os
import requests
import time

def fetch_paginated_data(base_url, api_key, max_pages=10):
    all_data = []
    page = 1
    
    while page <= max_pages:
        response = requests.get(
            base_url,
            params={'page': page, 'per_page': 100},
            headers={'Authorization': f'Bearer {api_key}'}
        )
        
        if response.status_code != 200:
            print(f"Error on page {page}: {response.status_code}")
            break
        
        data = response.json()
        if not data.get('items'):
            break
        
        all_data.extend(data['items'])
        print(f"Fetched page {page}: {len(data['items'])} items")
        
        if not data.get('has_next'):
            break
        
        page += 1
        time.sleep(0.5)
    
    return all_data

result = fetch_paginated_data(api_url, os.environ['SECRET_EXAMPLE_API_TOKEN'])
print(f"Total items fetched: {len(result)}")
""",
    variables={'api_url': 'https://api.example.com/data'},
    secrets=['secret_example_api_token'],
    return_var='result',
    timeout=120
)

​

Troubleshooting

​

Security Considerations

​

Sandbox Protections

• Restricted Imports: Blocks process-launch and native/FFI escape paths in local execution, including common modules used for host probing
• File System Isolation: Reads and writes are confined to BOT_GIT_PATH, BOT_REPOS_PATH, and BOT_SCRATCH_PATH; task-owned documents and data may be read, but source-code and sensitive-file reads are blocked even under those roots. Host paths outside those roots are refused. Responses sanitize absolute workspace paths to BOT_*_PATH shorthands
• Process Blocking: Local execution blocks subprocesses, forks, shell launches, and related process-creation APIs
• Native Library Blocking: Local execution blocks ctypes, cffi, and similar native loading paths
• Network Filtering: Only HTTP/HTTPS; no raw sockets or other protocols
• Resource Limits: CPU time and memory caps prevent denial-of-service
• Namespace Isolation: Each execution has isolated variable scope
• No Privilege Escalation: Sensitive shell/user environment variables are scrubbed before privilege-drop execution, and the local hardening is defense in depth rather than full kernel-enforced isolation

​

Safe Coding Practices

​

Performance Optimization

​

Efficient Data Processing

# Inefficient - Row-by-row iteration
python_exec(
    code="""
result = []
for row in data:
    result.append(row['value'] * 2)
"""
)

# Efficient - Vectorized operations
python_exec(
    code="""
import pandas as pd
df = pd.DataFrame(data)
result = (df['value'] * 2).tolist()
"""
)

​

Memory Management

python_exec(
    code="""
import pandas as pd

chunks = []
for chunk in pd.read_csv('large_file.csv', chunksize=10000):
    processed = chunk[chunk['amount'] > 100]
    chunks.append(processed)

result = pd.concat(chunks, ignore_index=True)
del chunks
"""
)