From c6beb650c4a35d30d3423106f579296ec9204729 Mon Sep 17 00:00:00 2001 From: "Abiba (pi)" Date: Sat, 27 Jun 2026 18:04:00 +0000 Subject: [PATCH] feat(agent-zero): Zulip adapter with A2A protocol for kagentz MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New project: agent-zero-zulip/ — direct Zulip integration for Agent Zero agents using A2A protocol internally. Architecture: kagentz (CT 105) ┌──────────────────────┐ │ Zulip Adapter ◄──► │ Agent Zero │ polls Zulip A2A │ Docker container │ port 8001 │ └──────────────────────┘ - adapter.py: Zulip event queue poller + A2A client - No intermediate bridge — direct kagentz ↔ Zulip - Heartbeat, reconnection, placeholder→edit streaming - Designed for re-use: scottdenya, future Agent Zero agents --- agent-zero-zulip/DEPLOY.md | 48 ++ agent-zero-zulip/README.md | 30 ++ agent-zero-zulip/config.yaml.example | 13 + agent-zero-zulip/requirements.txt | 1 + .../src/agent_zero_zulip/adapter.py | 485 ++++++++++++++++++ 5 files changed, 577 insertions(+) create mode 100644 agent-zero-zulip/DEPLOY.md create mode 100644 agent-zero-zulip/README.md create mode 100644 agent-zero-zulip/config.yaml.example create mode 100644 agent-zero-zulip/requirements.txt create mode 100644 agent-zero-zulip/src/agent_zero_zulip/adapter.py diff --git a/agent-zero-zulip/DEPLOY.md b/agent-zero-zulip/DEPLOY.md new file mode 100644 index 0000000..081eda6 --- /dev/null +++ b/agent-zero-zulip/DEPLOY.md @@ -0,0 +1,48 @@ +# Deploy Agent Zero Zulip Adapter + +## Prerequisites +- Agent Zero running in Docker on the target CT +- A Zulip bot created for the agent (kagentz-bot, scottdenya-bot, etc.) +- Python 3.11+ with httpx + +## Steps + +### 1. Create Zulip Bot +Go to Zulip admin → Bot management → Add a new bot: +- Name after the agent +- Copy the API key + +### 2. Install Adapter +```bash +# On the target CT +pip install httpx + +# Copy adapter files +scp -r agent-zero-zulip/ root@:/opt/agent-zero-zulip/ + +# Create config +cp config.yaml.example config.yaml +# Edit config.yaml with Zulip credentials +``` + +### 3. Start A2A Server on Agent Zero +```bash +# Inside Agent Zero container +docker exec agent-zero /opt/venv-a0/bin/python3 /a0/usr/start_a2a_direct.py & + +# Verify A2A is running +curl http://localhost:8001/.well-known/agent.json +``` + +### 4. Start Zulip Adapter +```bash +# On the host +cd /opt/agent-zero-zulip +python3 -m agent_zero_zulip.adapter & + +# Verify +tail -f adapter.log +``` + +### 5. Test +Send a DM to kagentz-bot in Zulip. Expect "Processing..." → response. diff --git a/agent-zero-zulip/README.md b/agent-zero-zulip/README.md new file mode 100644 index 0000000..994d3c5 --- /dev/null +++ b/agent-zero-zulip/README.md @@ -0,0 +1,30 @@ +# Agent Zero Zulip Adapter + +Direct Zulip integration for Agent Zero agents (kagentz, scottdenya, etc.). +Each agent gets its own Zulip bot. Uses A2A protocol internally to +communicate with Agent Zero's native A2A server. + +## Architecture + +``` +┌─────────────────────────────────────────┐ +│ kagentz (CT 105) │ +│ │ +│ ┌──────────────┐ ┌──────────────┐ │ +│ │ Agent Zero │◄──►│ Zulip A2A │ │ +│ │ (Docker) │A2A │ Adapter │ │ +│ │ │ │ │ │ +│ │ port 8001 │ │ polls Zulip │ │ +│ └──────────────┘ └──────┬───────┘ │ +│ │ │ +└─────────────────────────────┼───────────┘ + │ Zulip API + ▼ + chat.sysloggh.net +``` + +## Components + +- `adapter.py` — Zulip event queue poller + A2A client +- `agent_card.py` — Agent Card for A2A discovery +- `Dockerfile` or `requirements.txt` for dependencies diff --git a/agent-zero-zulip/config.yaml.example b/agent-zero-zulip/config.yaml.example new file mode 100644 index 0000000..30af244 --- /dev/null +++ b/agent-zero-zulip/config.yaml.example @@ -0,0 +1,13 @@ +# Agent Zero Zulip Adapter Configuration +# Copy to config.yaml and fill in credentials + +zulip: + site: https://chat.sysloggh.net + email: kagentz-bot@chat.sysloggh.net + api_key: "" + stream: agent-hub + +agent: + name: kagentz + a2a_url: http://localhost:8001/a2a + a2a_token: 8zNgdOEXzYxjQvTl diff --git a/agent-zero-zulip/requirements.txt b/agent-zero-zulip/requirements.txt new file mode 100644 index 0000000..aa69c38 --- /dev/null +++ b/agent-zero-zulip/requirements.txt @@ -0,0 +1 @@ +httpx>=0.27.0 diff --git a/agent-zero-zulip/src/agent_zero_zulip/adapter.py b/agent-zero-zulip/src/agent_zero_zulip/adapter.py new file mode 100644 index 0000000..c1d637b --- /dev/null +++ b/agent-zero-zulip/src/agent_zero_zulip/adapter.py @@ -0,0 +1,485 @@ +""" +Agent Zero Zulip Adapter — Direct Zulip integration for Agent Zero agents. + +Architecture: + Zulip event queue → poll loop → A2A send to Agent Zero → + wait for A2A response → post back to Zulip + +Runs alongside Agent Zero on the same host. Communicates with Agent Zero +via local A2A protocol (port 8001). No intermediate bridge needed. +""" + +import asyncio +import json +import logging +import os +import time +from datetime import datetime, timezone +from typing import Any, Optional +import urllib.parse + +try: + import httpx +except ImportError: + httpx = None + +logger = logging.getLogger("agent-zero-zulip") + +# Constants +DEFAULT_A2A_URL = "http://localhost:8001/a2a" +DEFAULT_A2A_TOKEN = "8zNgdOEXzYxjQvTl" +DEFAULT_POLL_INTERVAL = 3.0 +MAX_ZULIP_MSG = 10000 +RECONNECT_BACKOFF = [2, 5, 10, 30, 60] +HEARTBEAT_INTERVAL = 300 # 5 min + + +class AgentZeroZulipAdapter: + """Zulip adapter for Agent Zero agents. + + Connects to Zulip via event queue, forwards DMs and @mentions + to Agent Zero via local A2A protocol, and posts responses back. + """ + + def __init__(self): + # Zulip config from env vars + self._site = (os.getenv("ZULIP_SITE") or "").rstrip("/") + self._email = os.getenv("ZULIP_EMAIL") or "" + self._api_key = os.getenv("ZULIP_API_KEY") or "" + self._agent_name = os.getenv("ZULIP_AGENT_NAME") or "kagentz" + self._stream = os.getenv("ZULIP_STREAM") or "agent-hub" + + # A2A config + self._a2a_url = os.getenv("A2A_URL") or DEFAULT_A2A_URL + self._a2a_token = os.getenv("A2A_TOKEN") or DEFAULT_A2A_TOKEN + + # State + self._auth_header = self._build_auth() + self._http_client: Optional[httpx.AsyncClient] = None + self._queue_id: Optional[str] = None + self._last_event_id: int = -1 + self._bot_user_id: Optional[int] = None + self._bot_email: str = self._email + self._running = False + self._poll_task: Optional[asyncio.Task] = None + + # Stats + self._messages_processed = 0 + self._poll_count = 0 + self._reconnects = 0 + self._last_heartbeat = 0.0 + self._last_event_time = time.time() + self._a2a_sessions: dict = {} # chat_id -> A2A context_id + + def _build_auth(self) -> str: + import base64 + token = base64.b64encode(f"{self._email}:{self._api_key}".encode()).decode() + return f"Basic {token}" + + async def start(self): + """Start the adapter — connect to Zulip and begin polling.""" + if not all([self._site, self._email, self._api_key]): + logger.error("Missing Zulip credentials. Set ZULIP_SITE, ZULIP_EMAIL, ZULIP_API_KEY") + return False + + if not httpx: + logger.error("httpx not installed") + return False + + self._running = True + self._http_client = httpx.AsyncClient(timeout=30.0) + + # Connect to Zulip + if not await self._connect(): + return False + + logger.info(f"Adapter started for {self._agent_name} ({self._email})") + return True + + async def stop(self): + """Stop the adapter.""" + self._running = False + if self._poll_task: + self._poll_task.cancel() + try: + await self._poll_task + except asyncio.CancelledError: + pass + if self._http_client: + await self._http_client.aclose() + + async def _connect(self) -> bool: + """Register Zulip event queue.""" + try: + resp = await self._api_call("POST", "/api/v1/register", data={ + "event_types": '["message"]', + "apply_markdown": "true", + }) + if not resp: + logger.error("Queue registration failed") + return False + + self._queue_id = resp.get("queue_id") + self._last_event_id = resp.get("last_event_id", -1) + self._bot_user_id = resp.get("user_id") + + # Resolve bot user_id if not provided by register + if self._bot_user_id is None: + me = await self._api_call("GET", "/api/v1/users/me") + if me: + self._bot_user_id = me.get("user_id") + + logger.info( + f"Connected to Zulip as {self._email} " + f"(queue={self._queue_id}, bot_id={self._bot_user_id})" + ) + + # Start poll loop + self._poll_task = asyncio.create_task(self._poll_forever()) + return True + + except Exception as e: + logger.error(f"Connection failed: {e}") + return False + + async def _poll_forever(self): + """Main poll loop.""" + backoff = 0 + while self._running: + if not self._queue_id: + await asyncio.sleep(1) + continue + + try: + events = await self._fetch_events() + self._poll_count += 1 + + for event in events: + try: + await self._process_event(event) + except Exception as e: + logger.warning(f"Event processing error: {e}") + + backoff = 0 + + # Heartbeat + now = time.time() + if now - self._last_heartbeat > HEARTBEAT_INTERVAL: + self._last_heartbeat = now + silence = now - self._last_event_time + logger.info( + f"Heartbeat — polls={self._poll_count} " + f"processed={self._messages_processed} " + f"silence={silence:.0f}s reconnects={self._reconnects}" + ) + + except asyncio.CancelledError: + break + except Exception as e: + err = str(e) + if "BAD_EVENT_QUEUE_ID" in err: + logger.info("Queue expired, reconnecting...") + await self._reconnect() + continue + delay = RECONNECT_BACKOFF[min(backoff, len(RECONNECT_BACKOFF) - 1)] + backoff += 1 + logger.warning(f"Poll error: {e}") + await asyncio.sleep(delay) + + await asyncio.sleep(DEFAULT_POLL_INTERVAL) + + async def _fetch_events(self) -> list: + """Fetch events from Zulip queue.""" + resp = await self._api_call("GET", "/api/v1/events", params={ + "queue_id": self._queue_id, + "last_event_id": str(self._last_event_id), + "dont_block": "true", + }) + if resp is None: + return [] + + events = resp.get("events", []) + for event in events: + eid = event.get("id", 0) + if eid > self._last_event_id: + self._last_event_id = eid + + return [e for e in events if e.get("type") == "message"] + + async def _reconnect(self): + """Re-register event queue.""" + self._queue_id = None + try: + resp = await self._api_call("POST", "/api/v1/register", data={ + "event_types": '["message"]', + "apply_markdown": "true", + }) + if resp: + self._queue_id = resp.get("queue_id") + self._last_event_id = resp.get("last_event_id", -1) + self._reconnects += 1 + logger.info(f"Reconnected, queue={self._queue_id}") + except Exception as e: + logger.error(f"Reconnect failed: {e}") + + async def _process_event(self, event: dict): + """Process a Zulip message event.""" + msg = event.get("message", {}) + msg_type = msg.get("type", "") + sender_email = msg.get("sender_email", "") + sender_name = msg.get("sender_full_name", "Unknown") + sender_id = msg.get("sender_id") + content = msg.get("content", "") + + # Ignore own messages + if sender_email == self._bot_email: + return + + # Determine targeting + mentioned_users = msg.get("mentioned_users", []) or [] + mentioned_ids = [u.get("user_id") for u in mentioned_users if isinstance(u, dict)] + is_dm = msg_type == "private" + is_mention = self._bot_user_id and self._bot_user_id in mentioned_ids + + if is_dm: + self._messages_processed += 1 + self._last_event_time = time.time() + logger.info(f"DM from {sender_name}: {content[:80]}...") + + # Send typing indicator + await self._typing(sender_id, "start") + + # Send placeholder + placeholder_id = await self._send_msg("private", str(sender_id), + ":robot: _Processing your message..._") + + # Forward to Agent Zero via A2A + response = await self._a2a_chat(sender_name, content) + + # Stop typing + await self._typing(sender_id, "stop") + + # Post response + if response and placeholder_id: + await self._edit_msg(placeholder_id, response[:MAX_ZULIP_MSG]) + logger.info(f"Responded to {sender_name} ({len(response)} chars)") + elif response: + await self._send_msg("private", str(sender_id), response[:MAX_ZULIP_MSG]) + + async def _a2a_chat(self, sender_name: str, message: str) -> Optional[str]: + """Send message to Agent Zero via local A2A protocol.""" + try: + a2a_payload = { + "jsonrpc": "2.0", + "method": "tasks/send", + "params": { + "id": f"zulip-{int(time.time())}", + "message": { + "role": "user", + "parts": [{"text": f"[Zulip DM from {sender_name}]: {message}"}] + } + }, + "id": 1 + } + + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self._a2a_token}", + } + + async with httpx.AsyncClient(timeout=120.0) as client: + # Send the message to Agent Zero + resp = await client.post( + self._a2a_url, + json=a2a_payload, + headers=headers, + ) + + if resp.status_code != 200: + logger.error(f"A2A send failed: HTTP {resp.status_code}") + return f":warning: Failed to reach agent. Error: HTTP {resp.status_code}" + + result = resp.json() + task_id = result.get("result", {}).get("id") + + if not task_id: + logger.error("A2A: no task_id returned") + return ":warning: Agent did not create a task." + + # Poll for completion + for _ in range(60): # max 60s wait + await asyncio.sleep(1) + status_resp = await client.post( + self._a2a_url, + json={ + "jsonrpc": "2.0", + "method": "tasks/get", + "params": {"id": task_id}, + "id": 2 + }, + headers=headers, + ) + + if status_resp.status_code != 200: + continue + + status_data = status_resp.json() + result_data = status_data.get("result", {}) + state = result_data.get("status", {}).get("state", "") + + if state == "completed": + # Extract response text + return self._extract_a2a_response(result_data) + elif state in ("failed", "error"): + error_msg = result_data.get("status", {}).get("message", {}).get("text", "Unknown error") + return f":warning: Agent processing failed: {error_msg}" + + return ":hourglass: Agent did not complete in time." + + except httpx.TimeoutException: + return ":hourglass: Agent request timed out." + except Exception as e: + logger.error(f"A2A error: {e}") + return f":warning: Communication error: {type(e).__name__}" + + def _extract_a2a_response(self, result: dict) -> str: + """Extract assistant text from A2A response.""" + history = result.get("history", []) + for msg in reversed(history): + if isinstance(msg, dict) and msg.get("role") == "assistant": + parts = msg.get("parts", []) + texts = [p.get("text", "") for p in parts if isinstance(p, dict)] + text = "\n".join(t for t in texts if t) + if text: + return text + + artifacts = result.get("artifacts", []) + for artifact in reversed(artifacts): + if isinstance(artifact, dict): + text = artifact.get("text") or artifact.get("content") or "" + if text: + return text + + return "(no response)" + + # ── Zulip API ── + + async def _api_call(self, method: str, path: str, + data: dict = None, params: dict = None) -> Optional[dict]: + """Make Zulip API call.""" + if not self._http_client: + return None + + url = f"{self._site}{path}" + headers = {"Authorization": self._auth_header} + + try: + if method == "GET": + resp = await self._http_client.get(url, params=params, headers=headers) + elif method == "POST": + resp = await self._http_client.post(url, data=data, headers=headers) + elif method == "PATCH": + resp = await self._http_client.patch(url, data=data, headers=headers) + else: + return None + + if resp.status_code >= 400: + logger.debug(f"API {method} {path}: {resp.status_code}") + return None + + return resp.json() + except Exception as e: + logger.debug(f"API error {method} {path}: {e}") + return None + + async def _send_msg(self, msg_type: str, to: str, content: str) -> Optional[int]: + """Send a message to Zulip.""" + payload = {"type": msg_type, "content": content} + if msg_type == "private": + payload["to"] = json.dumps([int(to)]) + else: + payload["to"] = to + payload["subject"] = "general" + + resp = await self._api_call("POST", "/api/v1/messages", data=payload) + if resp and resp.get("id"): + return resp["id"] + return None + + async def _edit_msg(self, message_id: int, content: str): + """Edit a Zulip message.""" + await self._api_call("PATCH", f"/api/v1/messages/{message_id}", + data={"content": content}) + + async def _typing(self, user_id: int, op: str): + """Send typing indicator.""" + if not user_id: + return + to_data = json.dumps([user_id]) + await self._api_call("POST", "/api/v1/typing", + data={"to": to_data, "op": op}) + + def run(self): + """Synchronous entry point.""" + asyncio.run(self._run_async()) + + async def _run_async(self): + try: + if await self.start(): + # Keep running until stopped + while self._running: + await asyncio.sleep(1) + except asyncio.CancelledError: + pass + finally: + await self.stop() + + +# ── Entry point ── + +def main(): + """CLI entry point.""" + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + adapter = AgentZeroZulipAdapter() + + try: + adapter.run() + except KeyboardInterrupt: + logger.info("Shutdown requested") + + # Handle signals + import signal + shutdown = asyncio.Event() + + def _handle_signal(): + shutdown.set() + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + loop.add_signal_handler(signal.SIGTERM, _handle_signal) + loop.add_signal_handler(signal.SIGINT, _handle_signal) + except NotImplementedError: + pass + + async def _run(): + if await adapter.start(): + await shutdown.wait() + await adapter.stop() + + try: + loop.run_until_complete(_run()) + except KeyboardInterrupt: + pass + finally: + loop.close() + + +if __name__ == "__main__": + main()