Connectors

client.connectors (ConnectorsAPI) wraps FortiSOAR’s /api/integration surface – discovery, healthcheck, configuration, operation execution, the Connector Studio dev workspace, and install/uninstall. client.agents (AgentsAPI) covers the remote execution agent side: pushing, upgrading, and removing a connector on an agent, plus a liveness heartbeat.

A complete, runnable walkthrough lives in examples/manage_connectors.py – it defaults to read-only and exercises every method below.

Discovery & health

>>> client = demo_client()
>>> conn = client.connectors
>>> installed = conn.list_configured()           # installed + configured connectors
>>> [c.name for c in installed[:3]]
['smtp', 'code-snippet', ...]
>>> conn.resolve_version("mitre-attack")         # the configured version (None if absent)
'2.0.2'
>>> conn.resolve_version("not-installed") is None
True
>>> conn.configurations("mitre-attack")          # [{config_id, name, default}]
[ConnectorConfigSummary(id=7, config_id='01e4e6b4-c34e-4fc1-b692-bb08591f1fe5', name='Demo', default=True)]
>>> hc = conn.healthcheck("mitre-attack")        # status="Available" is green
>>> (hc.status, hc.name, hc.version)
('Available', 'mitre-attack', '2.0.2')

connector_detail fetches a connector’s full record – its operations (each with parameters + output_schema) and configurations. Captured live and trimmed to a doctest-friendly slice (the config dict on each configuration is dropped – it carries connection details):

>>> detail = conn.connector_detail("smtp")
>>> (detail["name"], detail["version"], detail["config_count"])
('smtp', '2.6.0', 1)
>>> [o["operation"] for o in detail["operations"][:3]]
['send_email_new', ...]
>>> [c["name"] for c in detail["configuration"]]
['localhost-postfix']

Executing an operation

execute() returns a typed ExecuteResult – .ok is the status == "Success" check, .data is the connector’s own output (shape varies by connector/operation). Live-verified against cisa-advisory’s get_known_exploited_vulnerability_cves – a public, read-only, parameter-less feed lookup safe to demo against a real vendor connector (the only side effect is CISA’s public catalog serving one GET):

>>> result = conn.execute("cisa-advisory", "get_known_exploited_vulnerability_cves")
>>> result.ok
True
>>> result.data["title"]
'CISA Catalog of Known Exploited Vulnerabilities'
>>> result.data["vulnerabilities"][0]["cveID"]
'CVE-2026-45659'

⚠️ For an agent-bound connector (see the module warning), execute() is fire-and-forget – it returns immediately with an in-progress status and empty data; the real result is pushed over a websocket, not pollable here.

Dynamic operation parameters (apiOperation)

A select/multiselect parameter can declare apiOperation – the name of a sibling connector operation whose result populates the dropdown at render time. This lets an operation’s choices come from the live remote system (VMs, severities, locations, …) instead of a hardcoded options list. The populating operation is visible: false (hidden from the playbook palette) and receives the connector config so it can authenticate.

action_ui_schema() returns the params with their apiOperation and apiOnchange fields, so a UI or agent can detect which params are dynamic:

params = conn.action_ui_schema("cisco-threatgrid", "submit_sample")
for p in params:
    if p.apiOperation:
        print(f"{p.name}: type={p.type} -> call {p.apiOperation}")
    else:
        print(f"{p.name}: type={p.type} (static)")

To resolve the choices for a dynamic param, call the populating operation via execute() – the result is a plain string list or [{"title": "...", "value": "..."}] objects:

defn = conn.definition("cisco-threatgrid")
op = next(o for o in defn.operations if o.operation == "submit_sample")
vm_param = next(p for p in op.parameters if p.apiOperation == "get_available_vms")

result = conn.execute("cisco-threatgrid", vm_param.apiOperation, config="<config-uuid>")
choices = result.data  # ["Windows 7 64-bit", "Linux 64-bit", ...]

When apiOnchange=True, the populating operation also receives the current values of all sibling parameters in params (for cascading dropdowns like sap-rfc’s pick-a-module-then-its-params-appear). Pass them as the params argument to execute:

result = conn.execute(
    "sap-rfc", "get_rfc_function_params",
    config="<config-uuid>",
    params={"function_name": "RFC_READ_TABLE"},  # sibling value
)
# result.data = {"options": "RFC_READ_TABLE", "onchange": {"RFC_READ_TABLE": [param, ...]}}

See the Connector Building Guide (section “Dynamic Options from an Operation”) for the full info.json declaration + Python handler patterns.

Creating, rotating, and deleting a configuration

create_configuration/update_configuration/delete_configuration write credentials via POST/PUT/DELETE /api/integration/configuration/. Captured live against a throwaway virustotal config (api_key is a placeholder value, never a real credential) – created, rotated, then deleted, leaving the box with 0 virustotal configs afterwards, same as before:

>>> created = conn.create_configuration(
...     "virustotal",
...     {"server": "www.virustotal.com", "api_key": "test-doctest-key", "verify_ssl": True},
...     name="pyfsr-doctest-config",
...     validate=False,   # skip the schema fetch (config_schema) for this offline demo
...     autofill=False,
... )
>>> (created.name, created.config["server"])
('pyfsr-doctest-config', 'www.virustotal.com')

Note config["api_key"] comes back as the literal string "NULL" regardless of what was sent – the server never echoes a stored secret, only this sentinel:

>>> created.config["api_key"]
'NULL'

update_configuration sends the config whole – include every field, not just the one you’re rotating:

>>> updated = conn.update_configuration(
...     "virustotal", created.config_id,
...     {"server": "www.virustotal.com", "api_key": "test-rotated-key", "verify_ssl": True},
...     name="pyfsr-doctest-config",
...     validate=False,
...     autofill=False,
... )
>>> updated.name
'pyfsr-doctest-config'

Note

The PUT response omits connector_name/connector_version – present on create_configuration’s response, absent on update_configuration’s. Don’t rely on either field being there after an update.

>>> conn.delete_configuration(created.config_id) is None
True

Making a configuration the default

A connector whose configuration isn’t marked default fails its healthcheck with Could not find a configuration matching the id get_default_config or the default configuration – the config is there and usable by name, but anything resolving by default gets nothing.

There is no flag-only route: PUT /api/integration/configuration/{config_id}/ replaces the whole record. That makes the obvious fix dangerous, because the listing returns config: null while only the single-record GET carries the real field map – build the PUT body from the listing and you wipe the credentials. set_default_configuration does the read-then-echo for you:

conn.set_default_configuration("fortigate-firewall")             # the only config
conn.set_default_configuration("fortigate-firewall", "a5fb56f2") # by config_id
conn.set_default_configuration("fortigate-firewall", name="fortigate-lab")

It deliberately skips validate/autofill – the stored config is already what the appliance accepted, and materializing it against the schema would rewrite fields this call has no business touching. A remote-agent binding is carried over explicitly; omitting agent on the PUT silently moves execution back to the self-agent. If the appliance returns the "NULL" secret sentinel instead of a stored value, the call raises rather than writing that sentinel over a live credential.

Note

The stored ciphertext for a secret legitimately changes across this call – the appliance re-encrypts on save while the plaintext does not change. Verified on a live 8.0.0 appliance: a FortiGate configuration that could not be health-checked reported Available afterwards, with the upstream still reachable. Don’t read the changed value as corruption.

Warning

The trailing slash on /api/integration/configuration/{config_id}/ is mandatory. Without it the gateway rejects the call with 403 Could not validate HMAC fingerprint, which reads like a permissions or auth problem rather than a URL typo.

Connector Studio dev workspace

Edit a checked-out connector’s source, then publish it onto the running appliance – the same flow as the in-product Studio editor.

dev = conn.dev_list()                       # connectors checked out for editing
entity_id = dev[0]["id"]

conn.dev_edit(entity_id)                     # open for editing (Studio "Edit")
conn.dev_read_file(entity_id, "/hello-world_1_0_0_dev/info.json")
conn.dev_write_file(entity_id, {"path": "info.json", "content": "{...}"})
conn.dev_publish(entity_id, replace=True)    # land changes + refresh integrations

Note

dev_publish() is also the supported escape hatch when a same-version .tgz upload left stale code cached in the integrations service – it triggers a service refresh the standard $replace=true install path does not.

Install / uninstall

# Appliance (self-agent):
conn.install("fortinet-fortisiem", "6.1.0", wait=True)   # by name from Content Hub
conn.install_from_file("hello-world-1.0.0.tgz", replace=True)  # upload a .tgz bundle
conn.uninstall("fortinet-fortisiem")

# Remote agent:
client.agents.install_connector(agent_id, name="cyops_utilities", version="3.7.1")
client.agents.upgrade_connector(agent_id, name="cyops_utilities", version="3.8.0")
client.agents.uninstall_connector(agent_id, name="cyops_utilities", version="3.8.0")

client.agents.heartbeat(agent_id)            # liveness over the secure-message bus

Warning

Appliance uninstall (uninstall()) and agent uninstall (uninstall_connector()) are distinct: the first removes the connector from the appliance’s self-agent by integer id, the second removes it from a named remote agent.