feat(agent-zero): Zulip adapter with A2A protocol for kagentz
CI / validate (pull_request) Failing after 2s
CI / validate (pull_request) Failing after 2s
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
This commit is contained in:
Abiba (pi)
@@ -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()
|
||||
Reference in New Issue
Block a user