echo.py

"""Echo agent — the canonical Agent Socket demo.

Connects as an agent and echoes every message it receives back to the
sender. Reads its token and address from a two-field JSON config::

    { "api_token": "sk_...", "agent_socket": "as:acme/echo" }

Run with::

    python echo.py config.json
"""
from __future__ import annotations

import json
import logging
import signal
import sys
from pathlib import Path

import agent_socket


def main() -> int:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")

    cfg_path = Path(sys.argv[1] if len(sys.argv) > 1 else "config.json")
    try:
        cfg = json.loads(cfg_path.read_text())
    except OSError as exc:
        print(f"read {cfg_path}: {exc}", file=sys.stderr)
        return 1
    except json.JSONDecodeError as exc:
        print(f"parse {cfg_path}: {exc}", file=sys.stderr)
        return 1

    token = cfg.get("api_token")
    socket_id = cfg.get("agent_socket")
    if not token or not socket_id:
        print(f"{cfg_path} must set api_token and agent_socket", file=sys.stderr)
        return 1

    def handle(m: agent_socket.Message) -> None:
        if m.err is not None:
            logging.error("error: %s", m.err)
            return
        logging.info("from %s: %s", m.from_, m.data)
        try:
            m.reply({"echo": m.data})
        except agent_socket.AgentSocketError as exc:
            logging.error("reply: %s", exc)

    agent = agent_socket.connect(token, socket_id, handle)

    def on_signal(_sig: int, _frame: object) -> None:
        agent.close()

    signal.signal(signal.SIGINT, on_signal)
    signal.signal(signal.SIGTERM, on_signal)

    logging.info("listening as %s (ctrl+c to quit)", socket_id)
    agent.wait()
    return 0


if __name__ == "__main__":
    sys.exit(main())