Skip to content

Event System

The Script Engine can subscribe to live audit events and query stored event history. Subscriptions are background tasks, so a script remains active until its tasks are stopped or the script itself is stopped.

Subscribing by event type

Callbacks receive the Event and a sequence of referenced objects. Even when an event normally references one object, handle the second argument as a sequence.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
import tuoni_script_engine as tse


def on_registered_agents(event, agents):
    for agent in agents:
        print(f"{event.time}: registered {agent.guid}")


task = tse.events.subscribe_by_event_type(
    tse.EventType.REGISTER_AGENT,
    on_registered_agents,
)

Callbacks may run asynchronously and should return promptly. If work can take a long time, avoid blocking unrelated callbacks with shared locks or mutable global state.

Subscribing by reference type

Subscribe to every event associated with a resource category:

1
2
3
4
5
6
7
8
9
def on_command_event(event, commands):
    for command in commands:
        print(event.type, command.id, command.status)


task = tse.events.subscribe_by_reference_type(
    tse.ReferenceType.COMMAND,
    on_command_event,
)

Event categories

Agent events

  • REGISTER_AGENT
  • DEACTIVATE_AGENT
  • REACTIVATE_AGENT
  • BLOCK_AGENT
  • UPDATE_AGENT_METADATA
  • MODIFY_AGENT_METADATA
  • AGENT_HEARTBEAT

Command events

  • CREATE_COMMAND
  • SEND_COMMAND
  • RECEIVE_COMMAND_RESULT
  • CREATE_COMMAND_UPDATE
  • SEND_COMMAND_UPDATE
  • CANCEL_COMMAND

Discovery events

  • CREATE_DISCOVERED_HOST, EDIT_DISCOVERED_HOST, ARCHIVE_DISCOVERED_HOST, RESTORE_DISCOVERED_HOST
  • CREATE_DISCOVERED_SERVICE, EDIT_DISCOVERED_SERVICE, ARCHIVE_DISCOVERED_SERVICE, RESTORE_DISCOVERED_SERVICE
  • CREATE_DISCOVERED_CREDENTIAL, EDIT_DISCOVERED_CREDENTIAL, ARCHIVE_DISCOVERED_CREDENTIAL, RESTORE_DISCOVERED_CREDENTIAL

Other events

  • CREATE_JOB, UPDATE_JOB
  • CHANGE_SETTINGS

Always pass enum members rather than raw strings:

1
2
3
4
task = tse.events.subscribe_by_event_type(
    tse.EventType.RECEIVE_COMMAND_RESULT,
    on_command_event,
)

Callback payload types

The subscription determines the resource sequence:

Event category or reference type Callback resources
Agent Sequence[Agent]
Command Sequence[Command]
Discovered host Sequence[DiscoveredHost]
Discovered service Sequence[DiscoveredService]
Discovered credential Sequence[DiscoveredCredential]
Job Sequence[Job]
Setting Sequence[Setting]

The Event object exposes:

Property Description
id Event UUID
type EventType
reference_type ReferenceType
reference_ids Identifiers referenced by the event
time Timezone-aware event timestamp

Subscription lifecycle

subscribe_by_event_type() and subscribe_by_reference_type() return a Task. The active task keeps the script alive.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
task = None


def stop_after_first_registration(event, agents):
    global task
    for agent in agents:
        print(f"First new agent: {agent.guid}")
    if task is not None:
        task.stop()


task = tse.events.subscribe_by_event_type(
    tse.EventType.REGISTER_AGENT,
    stop_after_first_registration,
)

A task is also stopped when its script file is changed or deleted, or when the Tuoni Server shuts down.

Querying event history

Get one event by UUID:

1
2
3
event = tse.events.get("550e8400-e29b-41d4-a716-446655440000")
if event is not None:
    print(event.type, event.time)

Iterate through events for a referenced object:

1
2
3
4
5
for event in tse.events.get_by_reference(
    tse.ReferenceType.AGENT,
    agent.guid,
):
    print(event.time, event.type)

Use the identifier type appropriate to the reference:

  • UUID or UUID string for agents and discovery records
  • integer for commands and jobs
  • string key for settings

Queueing a command from a callback

Use the global command manager because an event callback has no alias context:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
def collect_processes(event, agents):
    for agent in agents:
        command = tse.commands.queue_command(
            agent.guid,
            "ps",
            tse.JsonConfiguration({}),
        )
        command.wait_for_completion()

        if command.is_success() and command.result is not None:
            print(command.result.entries)
        elif command.result is not None:
            print(command.result.error_message or "Process listing failed")


task = tse.events.subscribe_by_event_type(
    tse.EventType.REGISTER_AGENT,
    collect_processes,
)

Long waits delay the callback. Use them only when that behavior is acceptable for the automation.

Updating an agent from a callback

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def mark_new_agents(event, agents):
    for agent in agents:
        agent.metadata.update_custom_properties({
            "registered_by_script": True,
            "registration_event": str(event.id),
        })


task = tse.events.subscribe_by_event_type(
    tse.EventType.REGISTER_AGENT,
    mark_new_agents,
)

Updating metadata emits further metadata events. Avoid subscribing to a metadata event and then unconditionally updating the same metadata, which can create an event loop.