Code

Code Tools

Build tools with custom JavaScript or Python code.

---

Overview

Code tools execute custom scripts to perform tasks that require:

• Complex business logic
• Dynamic data processing
• Custom calculations
• Conditional operations beyond visual flows
• Integration with systems requiring custom protocols

---

When to Use

Code tools are ideal when:

• Logic is dynamic or conditional
• You need fine-grained control over data handling
• Visual workflows become too complex
• External services require custom integration
• You need computational operations

Good Fit Examples

Use Case	Why Code Works
SQL query processor	Dynamic query construction based on user input
Custom validators	Complex validation rules with multiple conditions
Data transformers	Parse and reshape data from various formats
Algorithm execution	Mathematical calculations, sorting, filtering

---

Supported Languages

Language	Runtime	Best For
JavaScript	Node.js	Web APIs, JSON processing, async operations
Python	Python 3.x	Data processing, ML integration, scripting

---

Creating a Code Tool

Step 1: Create the Tool

1. Navigate to Tools → + New Tool
2. Select Code Tool
3. Choose language (JavaScript or Python)

Step 2: Define Interface

name: process_sql_query
description: |
  Executes dynamic SQL queries against the customer database.
  Converts natural language requests into SQL and returns results.

parameters:
  query_text:
    type: string
    description: The natural language query from the user
    required: true

  max_results:
    type: integer
    description: Maximum number of rows to return
    default: 100

Step 3: Write Code

JavaScript Example:

async function execute(input, context) {
  const { query_text, max_results = 100 } = input;

  // Parse the natural language query
  const sqlQuery = constructSQLFromText(query_text);

  // Execute against database
  const client = await getDBClient(context.env.DB_CONNECTION);

  try {
    const results = await client.query(sqlQuery, { limit: max_results });

    return {
      success: true,
      data: results.rows,
      count: results.rowCount,
      query: sqlQuery
    };
  } catch (error) {
    return {
      success: false,
      error: error.message,
      query: sqlQuery
    };
  }
}

function constructSQLFromText(text) {
  // Your SQL generation logic here
  // ...
}

Python Example:

def execute(input: dict, context: dict) -> dict:
    query_text = input.get("query_text")
    max_results = input.get("max_results", 100)

    # Parse the natural language query
    sql_query = construct_sql_from_text(query_text)

    # Execute against database
    connection = get_db_connection(context["env"]["DB_CONNECTION"])

    try:
        cursor = connection.cursor()
        cursor.execute(sql_query)
        results = cursor.fetchmany(max_results)

        return {
            "success": True,
            "data": results,
            "count": len(results),
            "query": sql_query
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e),
            "query": sql_query
        }


def construct_sql_from_text(text: str) -> str:
    # Your SQL generation logic here
    pass

Step 4: Test

1. Click Test
2. Provide sample input
3. Review output and execution logs
4. Debug any issues

Step 5: Deploy

Click Deploy to make the tool available to agents.

---

Code Structure

Entry Point

Your code must define an execute function:

JavaScript:

async function execute(input, context) {
  // Your code here
  return output;
}

Python:

def execute(input: dict, context: dict) -> dict:
    # Your code here
    return output

Input Parameter

Contains the parameters defined in your tool spec:

const input = {
  query_text: "Show orders from last week",
  max_results: 50
};

Context Object

Provides access to environment and runtime info:

const context = {
  env: {
    API_KEY: "sk-xxxx",
    DB_CONNECTION: "postgres://..."
  },
  session: {
    id: "session_123",
    user_id: "user_456"
  },
  memory: {
    // Session memory store
  }
};

Return Value

Return a JSON-serializable object:

return {
  status: "success",
  data: [...],
  metadata: {
    processed_at: new Date().toISOString()
  }
};

---

Environment Variables

Access secrets and configuration:

JavaScript:

const apiKey = context.env.API_KEY;
const dbUrl = context.env.DATABASE_URL;

Python:

api_key = context["env"]["API_KEY"]
db_url = context["env"]["DATABASE_URL"]

Setting variables:

1. Navigate to tool settings → Environment
2. Add key-value pairs
3. Mark sensitive values as secrets

---

HTTP Requests

JavaScript (fetch)

async function execute(input, context) {
  const response = await fetch('https://api.example.com/data', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${context.env.API_KEY}`
    },
    body: JSON.stringify(input)
  });

  const data = await response.json();
  return { success: true, data };
}

Python (requests)

import requests

def execute(input: dict, context: dict) -> dict:
    response = requests.post(
        'https://api.example.com/data',
        headers={
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {context["env"]["API_KEY"]}'
        },
        json=input
    )

    return {"success": True, "data": response.json()}

---

Error Handling

Always handle errors gracefully:

JavaScript:

async function execute(input, context) {
  try {
    const result = await riskyOperation(input);
    return { success: true, data: result };
  } catch (error) {
    console.error('Operation failed:', error);
    return {
      success: false,
      error: error.message,
      error_code: error.code || 'UNKNOWN'
    };
  }
}

Python:

def execute(input: dict, context: dict) -> dict:
    try:
        result = risky_operation(input)
        return {"success": True, "data": result}
    except Exception as e:
        print(f"Operation failed: {e}")
        return {
            "success": False,
            "error": str(e),
            "error_code": getattr(e, 'code', 'UNKNOWN')
        }

---

Examples

Data Validator

async function execute(input, context) {
  const { email, phone, date_of_birth } = input;
  const errors = [];

  // Validate email
  const emailRegex = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;
  if (!emailRegex.test(email)) {
    errors.push({ field: 'email', message: 'Invalid email format' });
  }

  // Validate phone
  const phoneRegex = /^\+?[\d\s-]{10,}$/;
  if (!phoneRegex.test(phone)) {
    errors.push({ field: 'phone', message: 'Invalid phone format' });
  }

  // Validate age
  const dob = new Date(date_of_birth);
  const age = Math.floor((Date.now() - dob) / (365.25 * 24 * 60 * 60 * 1000));
  if (age < 18) {
    errors.push({ field: 'date_of_birth', message: 'Must be 18 or older' });
  }

  return {
    valid: errors.length === 0,
    errors
  };
}

Price Calculator

def execute(input: dict, context: dict) -> dict:
    items = input.get("items", [])
    discount_code = input.get("discount_code")
    shipping_country = input.get("shipping_country", "US")

    # Calculate subtotal
    subtotal = sum(item["price"] * item["quantity"] for item in items)

    # Apply discount
    discount = 0
    if discount_code:
        discount = get_discount(discount_code, subtotal)

    # Calculate tax
    tax_rate = get_tax_rate(shipping_country)
    tax = (subtotal - discount) * tax_rate

    # Calculate shipping
    shipping = calculate_shipping(items, shipping_country)

    total = subtotal - discount + tax + shipping

    return {
        "subtotal": round(subtotal, 2),
        "discount": round(discount, 2),
        "tax": round(tax, 2),
        "shipping": round(shipping, 2),
        "total": round(total, 2)
    }

---

Execution Limits

Limit	Value
Timeout	30 seconds (configurable)
Memory	256 MB
Output size	1 MB

---

Debugging

Console Logging

JavaScript:

console.log('Input received:', input);
console.log('Processing step 1...');
console.error('Error occurred:', error);

Python:

print(f"Input received: {input}")
print("Processing step 1...")
print(f"Error occurred: {error}", file=sys.stderr)

Execution Logs

View logs in the tool’s monitoring section:

• Timestamp
• Log level
• Message content
• Execution context

---

Best Practices

Keep Functions Focused

One tool = one capability. Don’t combine unrelated logic.

Validate Input Early

if (!input.required_field) {
  return { error: 'required_field is required' };
}

Use Type Checking

def execute(input: dict, context: dict) -> dict:
    if not isinstance(input.get("amount"), (int, float)):
        return {"error": "amount must be a number"}

Handle Edge Cases

• Empty inputs
• Null values
• Invalid formats
• API failures

Document Your Code

Add comments for complex logic and business rules.

---

Related

• Tools Overview
• Workflow Tools
• MCP Tools
• Tool Calling