Playbook Authoring & Deployment¶

pyfsr can author FortiSOAR playbooks from YAML and deploy them through the same import path the UI uses. You write a collection as readable YAML, an optional compiler turns it into the FortiSOAR export envelope, and pyfsr pushes it to the appliance — no hand-building of workflow/step/route JSON.

The compiler is an optional extra¶

The YAML→JSON compiler ships separately from core pyfsr. Install it with the playbooks extra:

pip install "pyfsr[playbooks]"

Core pyfsr never imports it. Until it’s installed, the compile/deploy entry points raise PlaybooksExtraNotInstalled with that exact hint. The non-authoring collection methods (import_from_file(), list, delete, …) work without it.

Writing a playbook in YAML¶

A playbook file describes one collection and its workflows. The smallest useful shape (see the sample for the full file):

collection: pyfsr YAML Demo
description: Authored in YAML, deployed with pyfsr.
visible: true

playbooks:
  - name: pyfsr YAML Demo - Stamp Result
    is_active: false
    steps:
      - name: Start
        type: start
        next: Set Result

      - name: Set Result
        type: set_variable
        vars:
          greeting: hello from pyfsr
          source: yaml

The YAML schema (step types, their arguments, routing) is owned by the fsr_playbooks compiler. The compiler validates every step against a reference catalog of FortiSOAR step types and emits diagnostics (with code, path, message, and often a suggestion) when something won’t import.

Tip

For the full DSL — every top-level key, every step type, the friendly fields each accepts, and the start_on_create / start_on_update record triggers — see the Playbook YAML Syntax Reference.

Deploying from Python¶

The high-level path lives on client.workflow_collections. Compile and import in one call:

from pyfsr import FortiSOAR

client = FortiSOAR(base_url="https://fortisoar.example.com", auth="<api-key>")

created = client.workflow_collections.import_from_yaml(
    "alert_triage.yaml",
    replace=True,            # hard-delete + recreate a same-uuid collection
)
for col in created:
    print(col["name"], col["uuid"])

To inspect diagnostics before pushing anything, compile first (offline, no network) and check the result:

result = client.workflow_collections.compile_yaml("alert_triage.yaml")
if not result.ok:
    from pyfsr.authoring import format_diagnostic
    for diag in result.blocking:
        print(format_diagnostic(diag))
else:
    print("collections:", result.collection_names)
    print("playbooks:", result.playbook_names)
    client.workflow_collections.import_export(result.fsr_json, replace=True)

The CompiledPlaybook shape is doctested (compilation is offline, no network). ok is True only when there are no blocking errors; collection_names and playbook_names read off the produced envelope:

>>> from pyfsr.authoring import compile_playbook_yaml
>>> yaml = '''
... name: demo-triage
... description: Doctested example playbook
... playbooks:
...   - name: Triage Alert
...     description: one step
...     steps:
...       - name: Start
...         type: start
...         next: Set Note
...       - name: Set Note
...         type: set_variable
...         manual_input:
...           - name: note
...             type: text
...             value: hello
... '''
>>> result = compile_playbook_yaml(yaml)
>>> result.ok, result.collection_names, result.playbook_names
(True, ['00 - FSR Studio'], ['Triage Alert'])

A blocking error keeps ok False and leaves fsr_json None — errors holds every diagnostic so you can surface why before anything is deployed:

>>> bad = compile_playbook_yaml("name: x\nplaybooks:\n  - name: P\n    steps: []")
>>> bad.ok, bad.fsr_json
(False, None)
>>> [e["code"] for e in bad.errors]
['no_trigger']

import_from_yaml() options:

Option	Effect
`replace=True`	Hard-delete any existing collection whose uuid matches, then recreate (the UI’s “Replace existing playbook collection” flow). Without it a duplicate uuid raises `409 UniqueConstraintViolationException`.
`strict_warnings=True`	Treat compiler warnings as blocking, not just errors.
`db_path=...`	Override the reference catalog (defaults to the packaged one).

Compilation that produces blocking errors raises ValueError with the formatted diagnostics; a missing compiler raises PlaybooksExtraNotInstalled.

The compile result¶

compile_yaml() returns a CompiledPlaybook:

Attribute	Meaning
`ok`	True only when there are no blocking errors and an envelope was produced.
`fsr_json`	The `{"type": "workflow_collections", "data": [...]}` envelope, ready for `import_export` (`None` on blocking errors).
`errors`	Every diagnostic (errors and warnings) as dicts.
`blocking` / `warnings`	The error-only and warning-only subsets.
`collection_names` / `playbook_names`	Convenience name lists from the envelope.

Deploying from the CLI¶

The pyfsr playbook command group offers the same flow without writing Python. Unlike pyfsr appliance (which uses SSH), these talk to the FortiSOAR API and read connection details from the FSR_* environment (see EnvConfig), with optional flag overrides.

# Compile only — emit the envelope JSON, diagnostics to stderr (no network)
pyfsr playbook compile alert_triage.yaml -o envelope.json

# Validate — compile and report a diagnostics summary; nonzero exit on errors
pyfsr playbook validate alert_triage.yaml

# Deploy — compile then import via the API client
pyfsr playbook deploy alert_triage.yaml --replace

# See what deploy would create without posting anything
pyfsr playbook deploy alert_triage.yaml --dry-run

Connection overrides (any omitted value falls back to FSR_* env):

pyfsr playbook deploy alert_triage.yaml --replace \
    --server fortisoar.example.com --username csadmin --password '...' \
    --port 13002 --no-verify-ssl

Discovering step types¶

You don’t have to memorize the friendly type: keywords or their keys. The pyfsr playbook group is the authoring index — pyfsr playbook --help lists every affordance, and two offline commands enumerate the step catalog:

# List every friendly step type with its canonical FSR name + purpose
pyfsr playbook steps

# Keys, pitfalls, the typed-args schema, and a real compiling example for one type
pyfsr playbook step-help manual_input
pyfsr playbook step-help decision --schema

The same data is available from Python via pyfsr.playbook_catalog.list_step_types and pyfsr.playbook_catalog.step_help.

Worked examples — the foundational library¶

step-help shows one atom (one step type). The foundational playbook library shows whole molecules: complete, compiling, use-case-shaped playbooks you retrieve by intent and adapt — the layer an agent few-shots on when translating a goal to SOAR operations. It lives at examples/playbooks/library/, grouped by SOC stage (triggers / enrichment / decision / action / notify / control).

# List every library playbook with its stage, intent, step types, and compile status
pyfsr playbook examples

# Filter by intent or stage
pyfsr playbook examples --intent incident
pyfsr playbook examples --stage action

# Print one playbook's metadata + the full friendly YAML to adapt
pyfsr playbook show create-incident-from-alert

# Emit the retrieval manifest (intent + facets per playbook) for NL->playbook tooling
pyfsr playbook examples --manifest

Every library playbook compiles and carries a goal / trigger / inputs / outputs / connectors / adapts-to front-matter block. cold* in the compile column means it compiles but references connectors the offline slim catalog doesn’t carry — run pyfsr playbook deploy <file> --refresh-catalog to resolve them against a live instance. The manifest and listing are available from Python via pyfsr.playbook_library.list_library, pyfsr.playbook_library.library_manifest, and pyfsr.playbook_library.library_show (repo-internal, not part of the installed-package API).

Testing interactive playbooks & inspecting runs¶

A playbook that pauses on a Manual Input / Approval step can be driven end to end from Python. answer() finds the pending prompt, resolves the numeric run id / submit option / user, and resumes — in one call:

client.playbooks.trigger("Loop Until Six Digits")
# by_title matches the prompt's *schema title* — the step's `title:` — and NOT
# the step name, which here is "AskNumber":
client.manual_input.answer(654321, by_title="Enter a six digit number")

Note

A pending prompt’s .title is the manual_input step’s title:, mirrored from input.schema.title — not the step name. The two coincide only when the step declares no title:, in which case the schema title defaults to the step name.

Titles are also not unique: the same step paused in two runs yields two identically-titled rows, and answer() refuses an ambiguous match rather than resuming an arbitrary run. When you hold the trigger’s task_id, prefer the run-scoped pending_for_run(), whose rows carry the full prompt and the numeric run id.

To inspect what ran, run_tree() resolves a task_id to the run plus its child runs (no finding the parent by name), and step_status() reads a step’s outcome:

resp = client.playbooks.trigger("Loop Until Six Digits")
tree = client.playbooks.run_tree(resp["task_id"])   # parent + child runs, with pks
client.playbooks.step_status(tree.pk, "StampResult")  # -> "finished"

Note

FortiSOAR only records runtime set_variable / jinja values in the retrievable run record when global workflow debug logging is enabled; with it off (the default) run_env(...).env is empty for them. Either enable debug logging on the appliance to inspect env, or assert on a step’s status (via step_status), which is always recorded.

A complete worked example lives in examples/do_until_validation_loop.py.

Playbook version control (saved snapshots)¶

FortiSOAR keeps a snapshot history for each playbook — the editor’s “Versions” tab — backed by the workflow_versions module. (This is not a revision/diff resource; the word “revision” appears nowhere on the wire.) Each snapshot freezes the playbook definition at a point in time into a stringified json field, capped at 20 per playbook. pyfsr exposes it as list_versions() / get_version / create_version / restore_version / delete_version / diff_versions:

>>> client = demo_client()
>>> vers = client.playbooks.list_versions("Block IP (test fixture)")
>>> [v.note for v in vers]
['v2', 'v1']
>>> v1, v2 = vers[1], vers[0]
>>> v1.note, v1.autosave
('v1', False)

create_version captures the playbook as it is now (the server doesn’t echo the large json blob back on the POST — re-get_version to read it):

client.playbooks.create_version("Block IP", note="before-change")

diff_versions is client-side (FortiSOAR has no diff endpoint) — it compares two snapshots’ step graphs by uuid, surfacing added/removed/changed steps:

>>> d = client.playbooks.diff_versions(v1, v2)
>>> d.is_clean, [(c.step[:8], c.field) for c in d.changed]
(False, [('bbbb2222', 'arguments')])

restore_version overwrites the live playbook with a snapshot’s content (the editor’s flow: fetch the snapshot, parse its json, PUT it back) — call create_version first if you want a rollback point. It’s destructive, so the CLI gates it behind --yes:

client.playbooks.restore_version("Block IP", "<version-uuid>")

The CLI mirrors all six verbs under pyfsr playbook versions (list / get / create / restore / delete / diff):

pyfsr playbook versions list "Block IP"
pyfsr playbook versions diff <v1-uuid> <v2-uuid>
pyfsr playbook versions restore "Block IP" <version-uuid> --yes

Code-snippet sandbox: writing Python that runs¶

The code_snippet step runs a Python snippet through the code-snippet connector. The source goes under arguments.code: (a friendly shorthand the compiler maps to the canonical arguments.params.python_function). Two sandbox constraints shape every snippet you write — both confirmed against a live box — so they’re worth knowing up front:

The connector execs the snippet at module level. A top-level return is a SyntaxError. Put logic inside def functions and call them, or run statements inline — but never return from the top level of the snippet.
open (and the other filesystem builtins) are restricted. A snippet cannot read or write files. To produce a document a later step emails, embed the content inline in that step’s body: rather than writing a file the snippet can’t open.

Surfacing output: `print(json.dumps(...))` → `code_output`¶

There’s no return to receive a result. Instead, print a JSON document: the connector captures stdout and auto-deserializes the JSON into a structured code_output dict. Downstream steps read it at vars.steps.<name>.data.code_output.* — note data.code_output, not output.data, and it’s already a dict, so no | from_json filter is needed.

- name: Reconcile
  type: code_snippet
  arguments:
    code: |
      import json
      take = int("{{ vars.take }}" or 0)
      crew = int("{{ vars.crew_count }}" or 1)
      print(json.dumps({
          "cut_per_member": take // max(crew, 1),
          "risk": "high" if take > 1000000 else "low",
      }))
  next: Decide

- name: Decide
  type: decision
  conditions:
    - condition: "{{ vars.steps.Reconcile.data.code_output.risk == 'high' }}"
      label: big
      next: Alert
    - label: default
      next: Log

Reading upstream step output¶

A connector step’s result lives at vars.steps.<name>.data — that IS the step-result dict ({data, status, ...}); there is no separate .output level. For a code_snippet, .data.code_output is the deserialized dict; for a connector step, .data holds the operation’s response (e.g. vars.steps.FetchServiceNow.data.result). You can render these inline in the snippet as Python literals, which is how a code_snippet consumes upstream connector data without an API call:

- name: Diff
  type: code_snippet
  arguments:
    code: |
      import json
      a = {{ vars.steps.FetchFortiCloud.data.assets }}
      b = {{ vars.steps.FetchServiceNow.data.result }}
      # ...diff a vs b by join key...
      print(json.dumps({"findings": findings, "matched": matched}))

Note

A step’s display name with spaces is referenced in Jinja with the spaces replaced by underscores: a step named Apply Quarantine on FGT is vars.steps.Apply_Quarantine_on_FGT.status. Use a single-word step name (as above) to avoid the rewrite entirely.

Imports and the connector config¶

By default the sandbox restricts imports. To import from a package, the code-snippet connector’s configuration must allow it — set allow_imports on the config (true, or a restrictive list):

client.connectors.upsert_configuration(
    "code-snippet",
    {"allow_imports": True},
    name="default", default=True, validate=False,
)

default_config() shows the config schema for any connector (for code-snippet it surfaces allow_imports / restrict_imports). The whole step is also gated by the appliance’s Custom Code Execution system setting (set_custom_code_execution()) — with it off, a code_snippet step won’t run at all.

Tip

If a snippet genuinely needs a top-level return, unrestricted file access, or imports the sandbox forbids, escape the sandbox: run the Python through a custom unrestricted-python connector (a connector step pointing at, e.g., a code-runner connector) instead of the stock code_snippet step. The trade-off is a connector you must install and configure separately.

Importing an existing export¶

If you already have a *.json export from the UI’s Export button (no compiler needed), import it directly:

client.workflow_collections.import_from_file("exported_playbooks.json", replace=True)

Playbook Authoring & Deployment¶

The compiler is an optional extra¶

Writing a playbook in YAML¶

Deploying from Python¶

The compile result¶

Deploying from the CLI¶

Discovering step types¶

Worked examples — the foundational library¶

Testing interactive playbooks & inspecting runs¶

Playbook version control (saved snapshots)¶

Code-snippet sandbox: writing Python that runs¶

Surfacing output: print(json.dumps(...)) → code_output¶

Reading upstream step output¶

Imports and the connector config¶

Importing an existing export¶

Surfacing output: `print(json.dumps(...))` → `code_output`¶