| 1 | from __future__ import annotations |
| 2 | |
| 3 | import os |
| 4 | import shutil |
| 5 | import socket |
| 6 | from pathlib import Path |
| 7 | from typing import Any |
| 8 | from urllib.parse import urlparse |
| 9 | |
| 10 | from plugins._orchestrator.helpers.adapters.base import TerminalAgentAdapter |
| 11 | |
| 12 | # Inside the Agent Zero container, the local WebUI listens on port 80, |
| 13 | # so the default target is the instance running this very plugin. |
| 14 | DEFAULT_HOST = "http://localhost:80" |
| 15 | DEFAULT_DOCKER_A0_BINARY = "/opt/venv/bin/a0" |
| 16 | |
| 17 | |
| 18 | class AgentZeroAdapter(TerminalAgentAdapter): |
| 19 | """Delegate tasks to an Agent Zero instance via `a0 headless` one-shot mode. |
| 20 | |
| 21 | Host resolution: adapter config `host` > AGENT_ZERO_HOST env > local |
| 22 | instance (http://localhost:80 inside the container). |
| 23 | """ |
| 24 | |
| 25 | id = "a0" |
| 26 | title = "Agent Zero (headless)" |
| 27 | binary = "a0" |
| 28 | install_hint = "pip install git+https://github.com/agent0ai/a0-connector.git@development" |
| 29 | description = "Delegate to this or another Agent Zero instance through a0 headless." |
| 30 | |
| 31 | # --- connection ---------------------------------------------------------- |
| 32 | |
| 33 | def resolve_binary(self, config: dict[str, Any] | None = None) -> str: |
| 34 | binary = super().resolve_binary(config) |
| 35 | if binary == self.binary and shutil.which(binary) is None: |
| 36 | bundled = Path(DEFAULT_DOCKER_A0_BINARY) |
| 37 | if bundled.is_file(): |
| 38 | return str(bundled) |
| 39 | return binary |
| 40 | |
| 41 | def resolve_host(self, config: dict[str, Any] | None = None) -> str: |
| 42 | cfg = config or {} |
| 43 | host = str(cfg.get("host") or "").strip() |
| 44 | if host: |
| 45 | return host |
| 46 | env_host = os.environ.get("AGENT_ZERO_HOST", "").strip() |
| 47 | return env_host or DEFAULT_HOST |
| 48 | |
| 49 | def auth_status(self, config: dict[str, Any] | None = None) -> dict[str, Any]: |
| 50 | host = self.resolve_host(config) |
| 51 | if _probe(host): |
| 52 | return {"connected": True, "mode": "external", "auth_path": host} |
| 53 | return {"connected": False, "mode": "", "auth_path": host} |
| 54 | |
| 55 | |
| 56 | def _probe(host_url: str) -> bool: |
| 57 | url = host_url if "://" in host_url else f"http://{host_url}" |
| 58 | parsed = urlparse(url) |
| 59 | host = parsed.hostname or "localhost" |
| 60 | port = parsed.port or (443 if parsed.scheme == "https" else 80) |
| 61 | try: |
| 62 | with socket.create_connection((host, port), timeout=2): |
| 63 | return True |
| 64 | except OSError: |
| 65 | return False |