Skip to content

Code Steps Guide

Code steps execute Python in a secure sandbox. Use them when you need logic, transformation, or conditional behaviour between tool calls.

Basic syntax

steps:
  - id: process
    code: |
      # Your Python here.
      # Assign the step's output to the 'result' variable.
      data = context.inputs["data"]
      result = data.upper()

result = is required. The sandbox captures whatever you assign to result and makes it available to subsequent steps as steps.process.output. If you don't set result, the step returns None.

Accessing context

The sandbox injects a context object with three attributes:

context.inputs

Access workflow inputs by name:

url   = context.inputs["url"]           # raises KeyError if missing
limit = context.inputs.get("limit", 10) # safe with default

context.steps

Access output from any completed step:

response = context.steps["fetch"].output        # full output object
body     = context.steps["fetch"].output["body"] # nested field
count    = context.steps["fetch"].output.get("count", 0)

context.tools.call()

Call any MCP tool from within a code step:

response = await context.tools.call("http_request", {
    "url": context.inputs["url"],
    "method": "GET"
})
result = response.get("body", "")

Tool calls from code are rate-limited (default: 10 per step). Use tool steps for single calls; code steps for conditional or iterative logic.

context.tools.call_mcp()

Call a tool by MCP server name and tool name (avoids canonical runner__mcp__tool names):

commits = await context.tools.call_mcp("github", "list_commits", {
    "owner": "ostanlabs",
    "repo": "agent-execution-layer",
})
result = {"count": len(commits)}

Resolution priority: 1. CP-direct tool (registered on Control Plane) 2. defaults.runner from workflow YAML 3. Runner name from bridge context (X-Ploston-Runner header) 4. Single-match inference from RunnerRegistry

context.log()

Append a debug message to the step's log buffer. Log entries are captured in the step's debug_log field and available in templates as {{ steps.<id>.debug_log }}. Useful for tracing intermediate values without polluting the step output.

context.log(f"Processing {len(items)} items")
context.log(f"Filter matched: {matched}")
result = {"items": matched, "total": len(matched)}

context.workflow

Read-only workflow metadata:

wf_name = context.workflow.name          # workflow name
wf_ver  = context.workflow.version       # workflow version
exec_id = context.workflow.execution_id  # current execution ID
started = context.workflow.start_time    # ISO 8601 start time
result  = {"execution": exec_id}

Complete example

steps:
  - id: fetch
    tool: http_request
    params:
      url: "{{ inputs.url }}"
      method: GET

  - id: process
    depends_on: [fetch]
    code: |
      import re
      from datetime import datetime

      # Access previous step output
      body = context.steps["fetch"].output.get("body", "")

      # Extract title
      match = re.search(r"<title[^>]*>(.*?)</title>", body, re.IGNORECASE)
      title = match.group(1).strip() if match else "Unknown"

      # Build result
      result = {
          "title": title,
          "fetched_at": datetime.now().isoformat(),
          "length": len(body),
      }

Allowed imports

Module Use for
json JSON encoding/decoding
re Regular expressions
datetime Date and time handling
math Mathematical functions
random Random number generation
collections defaultdict, Counter, deque
itertools Iteration utilities
functools reduce, partial, lru_cache
hashlib Hashing
uuid UUID generation
decimal Decimal arithmetic
statistics mean, median, stdev
operator Standard operators
copy copy, deepcopy
time time.sleep, timestamps
typing Type hints
io io.BytesIO, io.StringIO
anthropic LLM synthesis steps (requires ANTHROPIC_API_KEY env var)
pypdf PDF parsing (e.g. pypdf.PdfReader)

Use imports at the top of the code block:

import json
import re
from datetime import datetime

data = json.loads(context.inputs["json_data"])
result = {"processed": True, "at": datetime.now().isoformat()}

Forbidden operations

The sandbox blocks anything that can escape the execution environment.

Forbidden builtins

eval, exec, compile, open, __import__, input, breakpoint, exit, quit, help, globals, locals, vars, dir, getattr, setattr, delattr, hasattr, callable, classmethod, staticmethod, property, super, type

Forbidden imports

os, sys, subprocess, socket, shutil, pathlib, urllib (full module — urllib.parse is allowed), requests, http, ctypes, pickle

Forbidden attribute access

Any dunder that enables class hierarchy traversal or code object inspection: __class__, __bases__, __base__, __mro__, __subclasses__, __code__, __globals__, __closure__, __func__, __builtins__, __dict__, __self__, __loader__, __spec__, __cached__, __file__, __path__

Error handling inside code

Use try/except to handle expected failures without failing the whole workflow:

steps:
  - id: safe_parse
    code: |
      import json

      try:
          data = json.loads(context.inputs["json_data"])
          result = {"ok": True, "data": data}
      except json.JSONDecodeError as e:
          result = {"ok": False, "error": str(e)}

A downstream step can then branch on steps.safe_parse.output["ok"].

Timeout

Default timeout is 30 seconds. Override per step:

steps:
  - id: slow_computation
    timeout: 120
    code: |
      # Up to 2 minutes
      result = expensive_calculation()

Common patterns

JSON processing

import json

raw  = context.inputs["json_string"]
data = json.loads(raw)
result = json.dumps(data, indent=2)

List transformation

items  = context.steps["fetch"].output["items"]
result = [{"id": i, "value": x * 2} for i, x in enumerate(items)]

Conditional logic

value = context.inputs["threshold"]
if value > 100:
    result = "high"
elif value > 50:
    result = "medium"
else:
    result = "low"

Aggregation

from collections import defaultdict

rows   = context.steps["query"].output["rows"]
groups = defaultdict(list)
for row in rows:
    groups[row["category"]].append(row["value"])

result = {k: sum(v) for k, v in groups.items()}

Calling a tool in a loop

import json

urls    = context.inputs["urls"]  # list of strings
results = []

for url in urls[:5]:  # respect rate limits — max 10 calls/step
    resp = await context.tools.call("http_request", {"url": url, "method": "GET"})
    results.append({"url": url, "status": resp.get("status_code")})

result = results

What the sandbox does NOT support

  • Reading or writing files — use fs_read / fs_write tool steps instead
  • Network access — use http_request tool steps instead
  • Spawning processes
  • Installing packages at runtime

All I/O goes through tool steps. Code steps handle transformation, logic, and aggregation.

Next steps