diff --git a/src/langbot/pkg/platform/sources/mattermost.py b/src/langbot/pkg/platform/sources/mattermost.py new file mode 100644 index 000000000..d9ae84c2c --- /dev/null +++ b/src/langbot/pkg/platform/sources/mattermost.py @@ -0,0 +1,375 @@ +from __future__ import annotations + +import asyncio +import json +import re +import typing +from urllib.parse import urlsplit, urlunsplit + +import aiohttp + +import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter +import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger +import langbot_plugin.api.entities.builtin.platform.entities as platform_entities +import langbot_plugin.api.entities.builtin.platform.events as platform_events +import langbot_plugin.api.entities.builtin.platform.message as platform_message + + +_MATTERMOST_MAX_POST_LENGTH = 16_383 +_MENTION_BOUNDARY = r'(? str: + """Return a validated Mattermost server URL without a trailing slash.""" + + url = server_url.strip().rstrip('/') + parsed = urlsplit(url) + if parsed.scheme not in {'http', 'https'} or not parsed.netloc: + raise ValueError('Mattermost server_url must be an absolute HTTP(S) URL') + return url + + +def _websocket_url(server_url: str) -> str: + parsed = urlsplit(server_url) + scheme = 'wss' if parsed.scheme == 'https' else 'ws' + return urlunsplit((scheme, parsed.netloc, f'{parsed.path}/api/v4/websocket', '', '')) + + +class MattermostMessageConverter(abstract_platform_adapter.AbstractMessageConverter): + """Translate Mattermost post text to and from LangBot message chains.""" + + @staticmethod + async def yiri2target(message_chain: platform_message.MessageChain) -> str: + parts: list[str] = [] + for component in message_chain: + if isinstance(component, platform_message.Plain): + parts.append(component.text) + elif isinstance(component, platform_message.Image) and component.url: + # Mattermost renders image URLs in Markdown messages. + parts.append(component.url) + elif isinstance(component, platform_message.File) and component.url: + parts.append(component.url) + return ''.join(parts) + + @staticmethod + async def target2yiri(post: dict, bot_username: str) -> platform_message.MessageChain: + text = str(post.get('message') or '') + components: list[typing.Any] = [ + platform_message.Source( + id=str(post.get('id') or ''), + time=float(post.get('create_at') or 0) / 1000, + ) + ] + if bot_username: + mention_pattern = re.compile(_MENTION_BOUNDARY.format(username=re.escape(bot_username)), re.IGNORECASE) + if mention_pattern.search(text): + components.append(platform_message.At(target=bot_username)) + text = mention_pattern.sub('', text).strip() + if text: + components.append(platform_message.Plain(text=text)) + return platform_message.MessageChain(components) + + +class MattermostEventConverter(abstract_platform_adapter.AbstractEventConverter): + @staticmethod + async def yiri2target(event: platform_events.MessageEvent) -> dict: + return event.source_platform_object + + @staticmethod + async def target2yiri( + post: dict, + channel: dict, + sender_name: str, + bot_username: str, + ) -> platform_events.MessageEvent: + message_chain = await MattermostMessageConverter.target2yiri(post, bot_username) + timestamp = float(post.get('create_at') or 0) / 1000 + sender_id = str(post.get('user_id') or '') + channel_type = channel.get('type') + + if channel_type == 'D': + return platform_events.FriendMessage( + sender=platform_entities.Friend(id=sender_id, nickname=sender_name or sender_id, remark=''), + message_chain=message_chain, + time=timestamp, + source_platform_object={'post': post, 'channel': channel}, + ) + + return platform_events.GroupMessage( + sender=platform_entities.GroupMember( + id=sender_id, + member_name=sender_name or sender_id, + permission=platform_entities.Permission.Member, + group=platform_entities.Group( + id=str(post.get('channel_id') or ''), + name=str(channel.get('display_name') or channel.get('name') or post.get('channel_id') or ''), + permission=platform_entities.Permission.Member, + ), + special_title='', + ), + message_chain=message_chain, + time=timestamp, + source_platform_object={'post': post, 'channel': channel}, + ) + + +class MattermostAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): + """Mattermost Bot Account adapter using the v4 REST and WebSocket APIs.""" + + server_url: str = '' + access_token: str = '' + session: aiohttp.ClientSession | None = None + listeners: dict[typing.Type[platform_events.Event], typing.Callable] = {} + channel_cache: dict[str, dict] = {} + stream_post_ids: dict[str, str] = {} + bot_username: str = '' + _running: bool = False + + message_converter: MattermostMessageConverter = MattermostMessageConverter() + event_converter: MattermostEventConverter = MattermostEventConverter() + + def __init__(self, config: dict, logger: abstract_platform_logger.AbstractEventLogger): + server_url = _normalize_server_url(str(config.get('server_url') or '')) + access_token = str(config.get('access_token') or '').strip() + if not access_token: + raise ValueError('Mattermost adapter requires an access_token') + + super().__init__( + config=config, + logger=logger, + server_url=server_url, + access_token=access_token, + bot_account_id='', + session=None, + listeners={}, + channel_cache={}, + stream_post_ids={}, + bot_username='', + _running=False, + ) + + async def _get_session(self) -> aiohttp.ClientSession: + if self.session is None or self.session.closed: + self.session = aiohttp.ClientSession( + headers={'Authorization': f'Bearer {self.access_token}'}, + raise_for_status=False, + ) + return self.session + + async def _api_request( + self, + method: str, + path: str, + *, + payload: dict | None = None, + ) -> dict: + session = await self._get_session() + async with session.request(method, f'{self.server_url}/api/v4{path}', json=payload) as response: + raw_body = await response.text() + if response.status >= 400: + # Mattermost returns a useful JSON error, but never include request headers/tokens in errors. + try: + error = json.loads(raw_body).get('message', raw_body) + except json.JSONDecodeError: + error = raw_body + raise RuntimeError(f'Mattermost API {method} {path} failed ({response.status}): {error}') + if not raw_body: + return {} + return json.loads(raw_body) + + async def _load_bot_identity(self) -> None: + user = await self._api_request('GET', '/users/me') + self.bot_account_id = str(user.get('id') or '') + self.bot_username = str(user.get('username') or '') + if not self.bot_account_id: + raise RuntimeError('Mattermost API did not return a bot user ID') + + async def _get_channel(self, channel_id: str) -> dict: + if channel_id not in self.channel_cache: + self.channel_cache[channel_id] = await self._api_request('GET', f'/channels/{channel_id}') + return self.channel_cache[channel_id] + + async def _post_message(self, channel_id: str, text: str, root_id: str = '') -> dict: + if not text: + return {} + if len(text) > _MATTERMOST_MAX_POST_LENGTH: + raise ValueError(f'Mattermost messages cannot exceed {_MATTERMOST_MAX_POST_LENGTH} characters') + payload = {'channel_id': channel_id, 'message': text} + if root_id: + payload['root_id'] = root_id + return await self._api_request('POST', '/posts', payload=payload) + + async def _get_direct_channel_id(self, user_id: str) -> str: + if not self.bot_account_id: + await self._load_bot_identity() + channel = await self._api_request( + 'POST', + '/channels/direct', + payload={'user_ids': [self.bot_account_id, user_id]}, + ) + channel_id = str(channel.get('id') or '') + if not channel_id: + raise RuntimeError('Mattermost did not return a direct-message channel ID') + self.channel_cache[channel_id] = channel + return channel_id + + async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain): + if target_type not in {'person', 'group'}: + raise ValueError("Mattermost target_type must be 'person' or 'group'") + text = await self.message_converter.yiri2target(message) + channel_id = str(target_id) + if target_type == 'person': + channel_id = await self._get_direct_channel_id(channel_id) + await self._post_message(channel_id, text) + + async def reply_message( + self, + message_source: platform_events.MessageEvent, + message: platform_message.MessageChain, + quote_origin: bool = False, + ): + source = await self.event_converter.yiri2target(message_source) + post = source['post'] + text = await self.message_converter.yiri2target(message) + # A message received inside a Mattermost thread must remain in that thread. When + # quote_origin is requested, make the response a reply to the source root post. + root_id = str(post.get('root_id') or '') + if quote_origin and not root_id: + root_id = str(post.get('id') or '') + await self._post_message(str(post['channel_id']), text, root_id) + + async def create_message_card(self, message_id: str, event: platform_events.MessageEvent) -> bool: + source = await self.event_converter.yiri2target(event) + post = source['post'] + root_id = str(post.get('root_id') or post.get('id') or '') + created = await self._post_message(str(post['channel_id']), 'Thinking…', root_id) + if created.get('id'): + self.stream_post_ids[str(message_id)] = str(created['id']) + return True + return False + + async def reply_message_chunk( + self, + message_source: platform_events.MessageEvent, + bot_message, + message: platform_message.MessageChain, + quote_origin: bool = False, + is_final: bool = False, + ): + response_id = str(bot_message.resp_message_id) + text = await self.message_converter.yiri2target(message) + if not text: + return + + post_id = self.stream_post_ids.get(response_id) + if post_id: + await self._api_request('PUT', f'/posts/{post_id}', payload={'id': post_id, 'message': text}) + else: + source = await self.event_converter.yiri2target(message_source) + post = source['post'] + root_id = str(post.get('root_id') or '') + if quote_origin and not root_id: + root_id = str(post.get('id') or '') + created = await self._post_message(str(post['channel_id']), text, root_id) + post_id = str(created.get('id') or '') + if post_id: + self.stream_post_ids[response_id] = post_id + + if is_final and getattr(bot_message, 'tool_calls', None) is None: + self.stream_post_ids.pop(response_id, None) + + async def is_stream_output_supported(self) -> bool: + return bool(self.config.get('enable_stream_reply', True)) + + def register_listener( + self, + event_type: typing.Type[platform_events.Event], + callback: typing.Callable[ + [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], typing.Awaitable[None] + ], + ): + self.listeners[event_type] = callback + + def unregister_listener( + self, + event_type: typing.Type[platform_events.Event], + callback: typing.Callable[ + [platform_events.Event, abstract_platform_adapter.AbstractMessagePlatformAdapter], typing.Awaitable[None] + ], + ): + self.listeners.pop(event_type, None) + + async def _dispatch_post(self, payload: dict) -> None: + data = payload.get('data') or {} + try: + post = json.loads(data.get('post') or '{}') + except (TypeError, json.JSONDecodeError): + await self.logger.error('Mattermost received a posted event with an invalid post payload') + return + + if not post or str(post.get('user_id') or '') == self.bot_account_id: + return + channel_id = str(post.get('channel_id') or '') + if not channel_id: + return + + try: + channel = await self._get_channel(channel_id) + event = await self.event_converter.target2yiri( + post, + channel, + str(data.get('sender_name') or post.get('user_id') or ''), + self.bot_username, + ) + callback = self.listeners.get(type(event)) + if callback: + result = callback(event, self) + if asyncio.iscoroutine(result): + await result + except Exception as exc: + await self.logger.error(f'Error handling Mattermost post: {exc}') + + async def _run_websocket_once(self) -> None: + session = await self._get_session() + async with session.ws_connect(_websocket_url(self.server_url), heartbeat=30) as websocket: + await websocket.send_json( + { + 'seq': 1, + 'action': 'authentication_challenge', + 'data': {'token': self.access_token}, + } + ) + async for message in websocket: + if message.type == aiohttp.WSMsgType.TEXT: + try: + payload = json.loads(message.data) + except json.JSONDecodeError: + continue + if payload.get('event') == 'posted': + await self._dispatch_post(payload) + elif message.type in {aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.ERROR}: + break + + async def run_async(self): + self._running = True + await self._load_bot_identity() + await self.logger.info(f'Mattermost bot connected: @{self.bot_username} ({self.bot_account_id})') + + retry_delay = 1 + while self._running: + try: + await self._run_websocket_once() + retry_delay = 1 + except asyncio.CancelledError: + raise + except Exception as exc: + if self._running: + await self.logger.error(f'Mattermost WebSocket disconnected: {exc}') + await asyncio.sleep(retry_delay) + retry_delay = min(retry_delay * 2, 30) + + async def kill(self) -> bool: + self._running = False + if self.session and not self.session.closed: + await self.session.close() + return True diff --git a/src/langbot/pkg/platform/sources/mattermost.svg b/src/langbot/pkg/platform/sources/mattermost.svg new file mode 100644 index 000000000..d185ef27d --- /dev/null +++ b/src/langbot/pkg/platform/sources/mattermost.svg @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/src/langbot/pkg/platform/sources/mattermost.yaml b/src/langbot/pkg/platform/sources/mattermost.yaml new file mode 100644 index 000000000..391f06902 --- /dev/null +++ b/src/langbot/pkg/platform/sources/mattermost.yaml @@ -0,0 +1,75 @@ +apiVersion: v1 +kind: MessagePlatformAdapter +metadata: + name: mattermost + label: + en_US: Mattermost + zh_Hans: Mattermost + zh_Hant: Mattermost + ja_JP: Mattermost + th_TH: Mattermost + vi_VN: Mattermost + es_ES: Mattermost + icon: mattermost.svg + description: + en_US: Mattermost Bot Account adapter using the v4 REST and WebSocket APIs. Add me to the teams and channels where you want me to interact. Please use a browser or desktop application to do this. + zh_Hans: 使用 Mattermost v4 REST API 与 WebSocket 的 Bot Account 适配器。请将我添加到您想要我互动的团队与频道。请使用浏览器或桌面应用进行操作。 + zh_Hant: 使用 Mattermost v4 REST API 與 WebSocket 的 Bot Account 介面卡。請將我加入您希望我互動的團隊與頻道。請使用瀏覽器或桌面應用程式操作。 + ja_JP: Mattermost v4 REST API と WebSocket を使用する Bot Account アダプター。利用させたいチームとチャンネルに私を追加してください。ブラウザまたはデスクトップアプリで操作してください。 + th_TH: อะแดปเตอร์ Bot Account ของ Mattermost ผ่าน v4 REST API และ WebSocket โปรดเพิ่มฉันไปยังทีมและช่องที่คุณต้องการให้ฉันโต้ตอบ โปรดดำเนินการผ่านเบราว์เซอร์หรือแอปเดสก์ท็อป + vi_VN: Bộ điều hợp Bot Account Mattermost sử dụng REST API v4 và WebSocket. Hãy thêm tôi vào các nhóm và kênh mà bạn muốn tôi tương tác. Vui lòng thao tác bằng trình duyệt hoặc ứng dụng máy tính để bàn. + es_ES: Adaptador de Bot Account de Mattermost mediante REST API v4 y WebSocket. Añádeme a los equipos y canales en los que quieras que interactúe. Hazlo desde un navegador o la aplicación de escritorio. +spec: + categories: + - global + - popular + config: + - name: server_url + label: + en_US: Mattermost Server URL + zh_Hans: Mattermost 服务器地址 + zh_Hant: 位址伺服器 Mattermost + ja_JP: Mattermost サーバー URL + th_TH: URL เซิร์ฟเวอร์ Mattermost + vi_VN: URL máy chủ Mattermost + es_ES: URL del servidor Mattermost + description: + en_US: The base URL of the Mattermost server, for example https://mattermost.example.com + zh_Hans: Mattermost 服务器基础地址，例如 https://mattermost.example.com + type: string + required: true + default: "" + - name: access_token + label: + en_US: Bot Access Token + zh_Hans: Bot 访问令牌 + zh_Hant: Bot 存取權杖 + ja_JP: Bot アクセストークン + th_TH: โทเค็นการเข้าถึงของบอต + vi_VN: Mã truy cập Bot + es_ES: Token de acceso del bot + description: + en_US: The personal access token generated for the Mattermost Bot Account + zh_Hans: 为 Mattermost Bot Account 生成的个人访问令牌 + type: string + required: true + default: "" + - name: enable_stream_reply + label: + en_US: Enable Stream Reply + zh_Hans: 启用流式回复 + zh_Hant: 啟用串流回覆 + ja_JP: ストリーミング返信を有効化 + th_TH: เปิดใช้งานการตอบกลับแบบสตรีม + vi_VN: Bật phản hồi luồng + es_ES: Activar respuesta en streaming + description: + en_US: Update a Mattermost post while LangBot generates a response + zh_Hans: 在 LangBot 生成回复时持续更新同一条 Mattermost 消息 + type: boolean + required: false + default: true +execution: + python: + path: ./mattermost.py + attr: MattermostAdapter diff --git a/tests/unit_tests/platform/test_mattermost_adapter.py b/tests/unit_tests/platform/test_mattermost_adapter.py new file mode 100644 index 000000000..6830ac6b6 --- /dev/null +++ b/tests/unit_tests/platform/test_mattermost_adapter.py @@ -0,0 +1,169 @@ +from types import SimpleNamespace + +import pytest + +from langbot.pkg.platform.sources.mattermost import ( + MattermostAdapter, + MattermostEventConverter, + MattermostMessageConverter, + _normalize_server_url, + _websocket_url, +) +import langbot_plugin.api.entities.builtin.platform.events as platform_events +import langbot_plugin.api.entities.builtin.platform.message as platform_message + + +class StubLogger: + async def info(self, *_args, **_kwargs): + pass + + async def error(self, *_args, **_kwargs): + pass + + +def _adapter() -> MattermostAdapter: + return MattermostAdapter.model_construct( + config={'enable_stream_reply': True}, + logger=StubLogger(), + server_url='https://mattermost.example.com', + access_token='secret', + bot_account_id='bot-id', + bot_username='langbot', + session=None, + listeners={}, + channel_cache={}, + stream_post_ids={}, + _running=False, + ) + + +def test_server_and_websocket_urls_preserve_subpath(): + server_url = _normalize_server_url('https://example.com/chat/') + assert server_url == 'https://example.com/chat' + assert _websocket_url(server_url) == 'wss://example.com/chat/api/v4/websocket' + + with pytest.raises(ValueError, match='absolute HTTP'): + _normalize_server_url('mattermost.example.com') + + +@pytest.mark.asyncio +async def test_converter_marks_and_removes_bot_mention(): + chain = await MattermostMessageConverter.target2yiri( + {'id': 'post-1', 'create_at': 1_000, 'message': '@langbot hello'}, + 'langbot', + ) + + assert any(isinstance(item, platform_message.At) for item in chain) + assert any(isinstance(item, platform_message.Plain) and item.text == 'hello' for item in chain) + + +@pytest.mark.asyncio +async def test_event_converter_distinguishes_direct_and_group_channels(): + post = {'id': 'post-1', 'channel_id': 'channel-1', 'user_id': 'user-1', 'message': 'hello', 'create_at': 1_000} + direct = await MattermostEventConverter.target2yiri(post, {'type': 'D'}, 'alice', 'langbot') + group = await MattermostEventConverter.target2yiri( + post, + {'type': 'O', 'display_name': 'General'}, + 'alice', + 'langbot', + ) + + assert isinstance(direct, platform_events.FriendMessage) + assert isinstance(group, platform_events.GroupMessage) + assert group.sender.group.name == 'General' + + +@pytest.mark.asyncio +async def test_send_to_person_creates_or_reuses_direct_channel(monkeypatch): + adapter = _adapter() + requests = [] + posted = [] + + async def api_request(method, path, *, payload=None): + requests.append((method, path, payload)) + return {'id': 'direct-channel', 'type': 'D'} + + async def post_message(channel_id, text, root_id=''): + posted.append((channel_id, text, root_id)) + return {'id': 'post-1'} + + monkeypatch.setattr(adapter, '_api_request', api_request) + monkeypatch.setattr(adapter, '_post_message', post_message) + + await adapter.send_message('person', 'user-1', platform_message.MessageChain([platform_message.Plain(text='hello')])) + + assert requests == [('POST', '/channels/direct', {'user_ids': ['bot-id', 'user-1']})] + assert posted == [('direct-channel', 'hello', '')] + + +@pytest.mark.asyncio +async def test_reply_keeps_existing_thread(monkeypatch): + adapter = _adapter() + posted = [] + + async def post_message(channel_id, text, root_id=''): + posted.append((channel_id, text, root_id)) + return {'id': 'reply'} + + monkeypatch.setattr(adapter, '_post_message', post_message) + event = platform_events.GroupMessage.model_construct( + source_platform_object={ + 'post': {'id': 'post-1', 'channel_id': 'channel-1', 'root_id': 'thread-root'}, + 'channel': {'type': 'O'}, + } + ) + + await adapter.reply_message(event, platform_message.MessageChain([platform_message.Plain(text='reply')])) + + assert posted == [('channel-1', 'reply', 'thread-root')] + + +@pytest.mark.asyncio +async def test_stream_reply_updates_existing_post(monkeypatch): + adapter = _adapter() + adapter.stream_post_ids['response-1'] = 'post-1' + requests = [] + + async def api_request(method, path, *, payload=None): + requests.append((method, path, payload)) + return {'id': 'post-1'} + + monkeypatch.setattr(adapter, '_api_request', api_request) + message = SimpleNamespace(resp_message_id='response-1', tool_calls=None) + + await adapter.reply_message_chunk( + SimpleNamespace(), + message, + platform_message.MessageChain([platform_message.Plain(text='complete')]), + is_final=True, + ) + + assert requests == [('PUT', '/posts/post-1', {'id': 'post-1', 'message': 'complete'})] + assert 'response-1' not in adapter.stream_post_ids + + +@pytest.mark.asyncio +async def test_posted_event_dispatches_listener(monkeypatch): + adapter = _adapter() + received = [] + + async def get_channel(_channel_id): + return {'type': 'D'} + + async def listener(event, _adapter): + received.append(event) + + monkeypatch.setattr(adapter, '_get_channel', get_channel) + adapter.register_listener(platform_events.FriendMessage, listener) + + await adapter._dispatch_post( + { + 'data': { + 'sender_name': 'alice', + 'post': '{"id":"post-1","channel_id":"channel-1","user_id":"user-1","message":"hello","create_at":1000}', + } + } + ) + + assert len(received) == 1 + assert received[0].sender.nickname == 'alice'