Python script

core.script.run_python

Execute a Python script.

Use this action to transform workflow data, import third-party packages, or call Tracecat APIs from an isolated Python 3.12 sandbox.

Script contract

Tracecat calls main if your script defines it; otherwise it calls the first non-private function.

Name the entrypoint main whenever your script contains more than one function, counting functions imported with from module import name. Validation rejects multi-function scripts without one.

Keep top-level import and def statements at the same indentation inside the script: block scalar.

Tracecat passes inputs to the entrypoint as keyword arguments, so keys must match the function’s parameter names. Give a parameter a default value when its key may be absent from inputs.

Set allow_network: true when your script makes outbound network requests. When the executor runs with TRACECAT__EXECUTOR_SANDBOX_ENABLED=true, deployment-wide CA certificates do not reach sandboxed scripts.

Return values

Tracecat serializes the function’s return value to JSON. Most types convert cleanly: datetime, date, and time become ISO 8601 strings (including inside lists and dicts), timedelta becomes seconds, set becomes a sorted list, Decimal, UUID, and Path become strings, an Enum becomes its value, a dataclass becomes an object, and bytes decodes as UTF-8 text.

Any other object becomes its repr() string, such as "<MyClass object at 0x...>". Tracecat raises Output not JSON-serializable only when serialization itself fails.

Import Python packages

Add pinned package specifications to dependencies, then import them in your script. Tracecat installs them in a separate phase that does not need allow_network: true.

- ref: parse_timestamp
  action: core.script.run_python
  args:
    inputs:
      timestamp: ${{ TRIGGER.created_at }}
    dependencies:
      - python-dateutil==2.9.0.post0
    script: |
      from dateutil import parser

      def main(timestamp):
          parsed = parser.isoparse(timestamp)
          return {
              "iso_timestamp": parsed.isoformat(),
              "timezone": str(parsed.tzinfo),
          }

Import Tracecat SDK

Import ctx from tracecat_registry; do not add the SDK to dependencies. Use clients such as ctx.cases, ctx.tables, ctx.variables, and ctx.workflows. Each client also provides an async variant under .aio, such as ctx.cases.aio.

SDK calls use the workflow’s internal authenticated execution context and do not require allow_network: true.

ctx.tables.insert_rows writes multiple rows. The destination table must already contain matching columns.

Synchronous

- ref: load_findings
  action: core.script.run_python
  args:
    inputs:
      findings: ${{ TRIGGER.findings }}
    script: |
      from tracecat_registry import ctx

      BATCH_SIZE = 500

      def transform_finding(finding):
          return {
              "finding_id": str(finding["id"]),
              "title": str(finding.get("title", "")).strip(),
              "severity": str(
                  finding.get("severity", "unknown")
              ).lower(),
              "observed_at": finding.get("observed_at")
              or finding.get("created_at"),
              "tags": sorted(set(finding.get("tags") or [])),
          }

      def main(findings):
          rows = [
              transform_finding(finding)
              for finding in findings
              if finding.get("id")
          ]

          rows_inserted = 0
          for start in range(0, len(rows), BATCH_SIZE):
              rows_inserted += ctx.tables.insert_rows(
                  table="findings",
                  rows_data=rows[start : start + BATCH_SIZE],
              )

          return {"rows_inserted": rows_inserted}

Asynchronous

- ref: load_findings_async
  action: core.script.run_python
  args:
    inputs:
      findings: ${{ TRIGGER.findings }}
    script: |
      from tracecat_registry import ctx

      BATCH_SIZE = 500

      def transform_finding(finding):
          return {
              "finding_id": str(finding["id"]),
              "title": str(finding.get("title", "")).strip(),
              "severity": str(
                  finding.get("severity", "unknown")
              ).lower(),
              "observed_at": finding.get("observed_at")
              or finding.get("created_at"),
              "tags": sorted(set(finding.get("tags") or [])),
          }

      async def main(findings):
          rows = [
              transform_finding(finding)
              for finding in findings
              if finding.get("id")
          ]

          rows_inserted = 0
          for start in range(0, len(rows), BATCH_SIZE):
              rows_inserted += await ctx.tables.aio.insert_rows(
                  table="findings",
                  rows_data=rows[start : start + BATCH_SIZE],
              )

          return {"rows_inserted": rows_inserted}

Inputs

script · string · required

Python script to execute. Must contain at least one function. If multiple functions are defined, one must be named ‘main’. Returns the output of the function.

allow_network · boolean

Whether to allow network access during script execution. Default is False. Set to True when the script makes external network requests. Dependency installation runs in a separate install phase and does not require it.

Default: false.

dependencies · array[string] | null

Optional list of Python package dependencies to install via pip. Packages are cached between executions for performance.

Default: null.

env_vars · map[string, string] | null

Environment variables to set in the sandbox. Use this to inject secrets or configuration.

Default: null.

inputs · object | null

Input data passed as keyword arguments to the main function. Keys must match the parameter names in the function signature; give a parameter a default value if its key may be missing.

Default: null.

timeout_seconds · integer

Maximum execution time in seconds. Default is 300 seconds (5 minutes).

Default: 300.

Examples

Enrich a payload

- ref: normalize_findings
  action: core.script.run_python
  args:
    inputs:
      findings: ${{ TRIGGER.findings }}
    script: |
      def main(findings):
          return [
              {
                  "id": finding["id"],
                  "severity": str(finding["severity"]).lower(),
              }
              for finding in findings
          ]
    timeout_seconds: 60