Event System

Scripts can register callbacks that fire when specific events occur on the Tuoni server. This enables reactive automation — auto-tagging agents on connect, logging command results, triggering follow-up actions, and more.

Registering Callbacks

import tuoni

def callback(event, data):
    print(f"Event: {event.type} at {event.time}")

task = tuoni.events.register_callback_by_event_type("REGISTER_AGENT", callback)

Parameters:

Parameter	Type	Description
`event_type`	`str`	One of the event type strings listed below
`callback`	`callable`	Function with signature `(event, data) -> None`

Returns: BackgroundTask — call task.stop() to unsubscribe.

Callback arguments:

Argument	Type	Description
`event`	Event	Event metadata (type, time, reference_type)
`data`	varies	Depends on the event's reference type (see table below)

Script Lifecycle

Registering a callback creates a background task that keeps the script alive. The script will not terminate until all background tasks are stopped. Call task.stop() to unsubscribe and allow the script to exit.

Event Types

Agent Events

Callback receives: Agent

Event Type	Description
`REGISTER_AGENT`	A new agent has connected to the server
`DEACTIVATE_AGENT`	An agent has been deactivated
`REACTIVATE_AGENT`	A previously deactivated agent has been reactivated
`BLOCK_AGENT`	An agent has been blocked
`UPDATE_AGENT_METADATA`	The agent reported updated metadata (e.g. new IPs, hostname change)
`MODIFY_AGENT_METADATA`	Agent metadata was modified by a user or script

import tuoni

def on_new_agent(event, agent):
    print(f"New agent: {agent.guid}")
    print(f"  Type: {agent.type}")
    print(f"  OS: {agent.metadata.os}")
    print(f"  Hostname: {agent.metadata.hostname}")
    print(f"  IPs: {agent.metadata.ips}")

task = tuoni.events.register_callback_by_event_type("REGISTER_AGENT", on_new_agent)

Command Events

Callback receives: CommandState

Event Type	Description
`CREATE_COMMAND`	A command has been created
`SEND_COMMAND`	A command has been sent to an agent
`RECEIVE_COMMAND_RESULT`	A result (partial or final) has been received for a command
`CREATE_COMMAND_UPDATE`	A command update has been created
`CANCEL_COMMAND`	A command has been cancelled

import tuoni

def on_result(event, cmd):
    if cmd.result is not None and cmd.result.is_finished():
        stdout = cmd.result.items.get("STDOUT", "")
        print(f"Command {cmd.id} on agent {cmd.agent_guid}: {stdout}")

task = tuoni.events.register_callback_by_event_type("RECEIVE_COMMAND_RESULT", on_result)

Partial Results

RECEIVE_COMMAND_RESULT fires for every result, including partial (ongoing) results. Check cmd.result.is_finished() to determine if the command has fully completed.

Discovery Events

Host Events

Callback receives: DetailedDiscoveredHost

Event Type	Description
`CREATE_DISCOVERED_HOST`	A new host was added to discovery
`EDIT_DISCOVERED_HOST`	A discovered host was modified
`ARCHIVE_DISCOVERED_HOST`	A discovered host was archived
`RESTORE_DISCOVERED_HOST`	An archived host was restored

Service Events

Callback receives: DetailedDiscoveredService

Event Type	Description
`CREATE_DISCOVERED_SERVICE`	A new service was added to discovery
`EDIT_DISCOVERED_SERVICE`	A discovered service was modified
`ARCHIVE_DISCOVERED_SERVICE`	A discovered service was archived
`RESTORE_DISCOVERED_SERVICE`	An archived service was restored

Credential Events

Callback receives: DetailedDiscoveredCredential

Event Type	Description
`CREATE_DISCOVERED_CREDENTIAL`	A new credential was added to discovery
`EDIT_DISCOVERED_CREDENTIAL`	A discovered credential was modified
`ARCHIVE_DISCOVERED_CREDENTIAL`	A discovered credential was archived
`RESTORE_DISCOVERED_CREDENTIAL`	An archived credential was restored

Job Events

Callback receives: JobState

Event Type	Description
`CREATE_JOB`	A new job (e.g. script execution) was created
`UPDATE_JOB`	A job's status or messages changed

Settings Events

Event Type	Description
`CHANGE_SETTINGS`	A server setting was changed

Patterns

One-Shot Listener

Listen for a single event, then stop:

import tuoni

def on_first_agent(event, agent):
    print(f"First agent connected: {agent.guid}")
    task.stop()  # Stop listening after the first event

task = tuoni.events.register_callback_by_event_type("REGISTER_AGENT", on_first_agent)

Variable Scope

The task variable must be accessible inside the callback. Define it at module scope (as shown above), not inside a function.

Long-Running Listener

Listen indefinitely — the script stays alive as long as the task is active:

import tuoni

def on_any_agent(event, agent):
    print(f"Agent event ({event.type}): {agent.guid}")

# This script runs forever (until the file is deleted or server stops)
task = tuoni.events.register_callback_by_event_type("REGISTER_AGENT", on_any_agent)
print("Listening for new agents...")

Event-Driven Command Execution

Combine events with direct command sending:

import tuoni

def on_new_agent(event, agent):
    print(f"New agent: {agent.guid}, sending hostname command...")
    cmd = tuoni.commands.queue_command(
        agent.guid,
        "sh",
        {"command": "echo Hello from script!"}
    )
    cmd.wait_for_finish()
    if cmd.is_success():
        print(f"Result: {cmd.get_result()['STDOUT']}")
    else:
        print(f"Failed: {cmd.get_error_message()}")

task = tuoni.events.register_callback_by_event_type("REGISTER_AGENT", on_new_agent)
print("Waiting for agents...")

Auto-Tagging Agents

Tag agents automatically based on their metadata when they connect:

import tuoni

def auto_tag(event, agent):
    ips = agent.metadata.ips or ""
    for ip in ips.split():
        octets = ip.split(".")
        if len(octets) == 4:
            try:
                subnet = int(octets[2])
                tag = f"#subnet-{subnet}"
                notes = agent.metadata.custom_properties.get("notes", "")
                if tag not in notes.split(";"):
                    agent.metadata.custom_properties["notes"] = (
                        f"{notes};{tag}" if notes else tag
                    )
                break
            except ValueError:
                continue

task = tuoni.events.register_callback_by_event_type("REGISTER_AGENT", auto_tag)
print("Auto-tagging enabled.")

Monitoring Command Results

Log all command results across all agents:

import tuoni

def log_result(event, cmd):
    if cmd.result is None:
        return
    status = cmd.result.status
    agent = cmd.agent_guid
    template = cmd.command_template_id
    print(f"[{status}] Agent {agent} - {template} (cmd #{cmd.id})")
    if cmd.result.is_finished() and cmd.result.items:
        for name, value in cmd.result.items.items():
            if isinstance(value, str):
                preview = value[:200]
                print(f"  {name}: {preview}")

task = tuoni.events.register_callback_by_event_type("RECEIVE_COMMAND_RESULT", log_result)
print("Command result monitor active.")