kagentz-bot
941b69e782
CI / validate (pull_request) Failing after 2s
- adapter.py: add import asyncio, use Client(site=) for Python 3.13 compat - adapter.py: use asyncio.new_event_loop() in background thread - adapter.py: add print() in _process_event for journald visibility - deploy.sh: refactor to PLATFORM-based maps instead of per-agent duplications - deploy.sh: remove backslash artifacts, deduplicate service/plugin path logic
138 lines
5.1 KiB
Python
138 lines
5.1 KiB
Python
# Core adapter implements BasePlatformAdapter for Zulip
|
|
# See ADR-005 for @mention detection, ADR-009 for error handling
|
|
# Full implementation in issue #5.
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
import re
|
|
import threading
|
|
from typing import Any, Dict, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class ZulipAdapter:
|
|
"""Zulip adapter for Hermes agents. Connects to Zulip, listens for
|
|
@mentions in #agent-hub, routes messages via BasePlatformAdapter."""
|
|
|
|
# Regex to strip Zulip mention artifacts from message bodies (ADR-008)
|
|
MENTION_CLEANER = re.compile(r'@\*\*[^*]+\*\*')
|
|
|
|
def __init__(self, config: Dict[str, Any]) -> None:
|
|
self.config = config
|
|
self.connected = False
|
|
self._client = None
|
|
self._thread = None
|
|
self._stop_event = threading.Event()
|
|
|
|
async def connect(self) -> None:
|
|
"""Establish connection to Zulip and start the event loop."""
|
|
if self.connected:
|
|
logger.info("Already connected to Zulip.")
|
|
return
|
|
|
|
try:
|
|
from zulip import Client
|
|
server_url = self.config['zulip']['server_url']
|
|
email = self.config['zulip']['email']
|
|
api_key = self.config['zulip']['api_key']
|
|
|
|
self._client = Client(site=server_url.rstrip('/'), email=email, api_key=api_key)
|
|
self.connected = True
|
|
logger.info(f"Connected to Zulip: {server_url} as {email}")
|
|
|
|
# Start the event loop in a background thread
|
|
self._stop_event.clear()
|
|
self._thread = threading.Thread(target=self._event_loop, daemon=True)
|
|
self._thread.start()
|
|
logger.info("Zulip event loop started.")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to connect to Zulip: {e}")
|
|
self.connected = False
|
|
raise
|
|
|
|
async def disconnect(self) -> None:
|
|
"""Disconnect from Zulip and stop the event loop."""
|
|
if not self.connected:
|
|
return
|
|
self._stop_event.set()
|
|
if self._thread:
|
|
self._thread.join(timeout=5)
|
|
self.connected = False
|
|
logger.info("Disconnected from Zulip.")
|
|
|
|
async def send_message(self, topic: str, content: str) -> Dict[str, Any]:
|
|
"""Send a message to Zulip with retry logic (ADR-009)."""
|
|
max_retries = self.config.get('error_handling', {}).get('retry_count', 3)
|
|
retry_delay = self.config.get('error_handling', {}).get('retry_delay_seconds', 5)
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
result = self._client.send_message({
|
|
'type': 'stream',
|
|
'to': self.config['zulip']['stream'],
|
|
'subject': topic,
|
|
'content': content,
|
|
})
|
|
logger.info(f"Message sent to {self.config['zulip']['stream']} > {topic} (attempt {attempt + 1})")
|
|
return result
|
|
except Exception as e:
|
|
logger.warning(f"Failed to send message (attempt {attempt + 1}): {e}")
|
|
if attempt < max_retries - 1:
|
|
time.sleep(retry_delay)
|
|
else:
|
|
raise
|
|
|
|
async def on_event(self, event: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""Process incoming Zulip events and route via BasePlatformAdapter."""
|
|
if event.get('type') != 'message':
|
|
return None
|
|
|
|
msg = event.get('message', {})
|
|
mentioned_users = msg.get('mentioned_users', False)
|
|
|
|
# Only process messages that mention this bot (ADR-005)
|
|
if not mentioned_users:
|
|
return None
|
|
|
|
# Strip Zulip formatting artifacts
|
|
body = msg.get('content', '')
|
|
clean_body = self.MENTION_CLEANER.sub('', body)
|
|
clean_body = clean_body.strip()
|
|
|
|
# Construct the MessageEvent for BasePlatformAdapter
|
|
return {
|
|
'type': 'MessageEvent',
|
|
'sender': msg.get('sender_full_name', 'Unknown'),
|
|
'body': clean_body,
|
|
'topic': msg.get('subject', 'Unknown Topic'),
|
|
'stream': msg.get('stream', 'Unknown Stream'),
|
|
'timestamp': msg.get('timestamp', 0),
|
|
}
|
|
|
|
def _event_loop(self) -> None:
|
|
# Create an event loop for this background thread (Python 3.13+)
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
"""Background thread to listen for Zulip events."""
|
|
try:
|
|
self._client.call_on_each_message(
|
|
lambda event: self._process_event(event),
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Event loop crashed: {e}")
|
|
self.connected = False
|
|
|
|
def _process_event(self, event: Dict[str, Any]) -> None:
|
|
"""Bridge the synchronous event to the async on_event handler."""
|
|
import asyncio
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
if loop.is_running():
|
|
asyncio.create_task(self.on_event(event))
|
|
else:
|
|
asyncio.run(self.on_event(event))
|
|
except Exception as e:
|
|
logger.error(f"Error processing event: {e}")
|