scripts.agent_notify

Reference › scripts › agent_notify

A short script that an agent runs inside its Incus container to tell the host that its task has finished. It is the sending half of the pair completed by agent_check on the host. Host and port come from resolve_redis_host and resolve_redis_port in redis_factory.

Invocation: --task is required. --status is done (default) or failed, --message is optional free text, and --redis-host, --redis-port and --redis-db override the connection (the database defaults to the REDIS_DB variable, else 0). It exits with an error message if the redis package is missing.

main builds a small JSON payload with the agent name, task, status and message. It publishes the payload on the channel agent:done:<task>, which wakes a listening agent_check, and also stores it under the key agent:status:<task> with a one-hour expiry, so a check started late can still read the result.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/agent_notify.py

#!/usr/bin/env python3
"""Agent notification script — run INSIDE the Incus container when task is done.

Sends a Redis message to the host listener with task status.
Usage (inside container):
    python agent_notify.py --task TASK_ID --status done --message "Ready for review"
"""

import argparse
import json
import os
import sys

from orchestrata_ultimate.core.redis_factory import (
    resolve_redis_host,
    resolve_redis_port,
)


def main() -> None:
    parser = argparse.ArgumentParser(description="Notify host that agent work is done")
    parser.add_argument("--task", required=True, help="Task identifier")
    parser.add_argument(
        "--status", default="done", choices=["done", "failed"], help="Task status"
    )
    parser.add_argument("--message", default="", help="Optional message")
    parser.add_argument("--redis-host", default=None)
    parser.add_argument("--redis-port", type=int, default=None)
    parser.add_argument(
        "--redis-db", type=int, default=int(os.environ.get("REDIS_DB", "0"))
    )
    args = parser.parse_args()

    try:
        import redis
    except ImportError:
        print(
            "ERROR: redis package not installed. Install with: uv add redis",
            file=sys.stderr,
        )
        sys.exit(1)

    client = redis.Redis(
        host=args.redis_host or resolve_redis_host(),
        port=args.redis_port or resolve_redis_port(),
        db=args.redis_db,
        decode_responses=True,
    )

    payload = {
        "agent": "claude",
        "task": args.task,
        "status": args.status,
        "message": args.message,
    }

    channel = f"agent:done:{args.task}"
    client.publish(channel, json.dumps(payload))
    client.setex(f"agent:status:{args.task}", 3600, json.dumps(payload))

    print(f"Notification sent on channel '{channel}': {payload}")


if __name__ == "__main__":
    main()