> ## Documentation Index
> Fetch the complete documentation index at: https://tracecat-codex-docs-secrets-oauth-discoverability.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.
Tracecat passes `inputs` as function arguments and returns the
function result as the action output.

### Import Python packages

Add pinned package specifications to `dependencies`, then import them
in your script. Set `allow_network: true` so Tracecat can install the
packages from PyPI. Tracecat caches installed packages between runs.

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

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

### Import Tracecat SDK

The Tracecat SDK is already available in the Python sandbox. 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`.

Use `ctx.tables.insert_rows` to write multiple rows in one request.
For large inputs, transform the records first and split them into
bounded batches. The destination table must already contain matching
columns.

<Tabs>
  <Tab title="Synchronous">
    ```yaml theme={null}
    - 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}
    ```
  </Tab>

  <Tab title="Asynchronous">
    ```yaml theme={null}
    - 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}
    ```
  </Tab>
</Tabs>

### Inputs

<ParamField path="script" type="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.
</ParamField>

<ParamField path="allow_network" type="boolean">
  Whether to allow network access during script execution. Default is False. Set to True when installing PyPI dependencies or making external network requests.

  Default: `false`.
</ParamField>

<ParamField path="dependencies" type="array[string] | null">
  Optional list of Python package dependencies to install via pip. Packages are cached between executions for performance.

  Default: `null`.
</ParamField>

<ParamField path="env_vars" type="map[string, string] | null">
  Environment variables to set in the sandbox. Use this to inject secrets or configuration.

  Default: `null`.
</ParamField>

<ParamField path="inputs" type="object | null">
  Input data passed as function arguments to the main function. Keys must match the parameter names in the function signature. Missing parameters will receive `None`.

  Default: `null`.
</ParamField>

<ParamField path="timeout_seconds" type="integer">
  Maximum execution time in seconds. Default is 300 seconds (5 minutes).

  Default: `300`.
</ParamField>

### Examples

**Enrich a payload**

```yaml theme={null}
- 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
```
