From 9fe0e65ae3c7c8083f1602a9804ecf6803608271 Mon Sep 17 00:00:00 2001 From: moe-berri Date: Mon, 12 Oct 2026 14:45:31 -0400 Subject: [PATCH 1/6] refactor: add bounded dispatch lifecycle for channel adapters --- src/agentchat/channels/_delivery.py | 59 +++++++++++++++++++++ tests/contracts/test_delivery.py | 82 +++++++++++++++++++++++++++++ tests/support/receipts.py | 11 ++++ 3 files changed, 152 insertions(+) create mode 100644 src/agentchat/channels/_delivery.py create mode 100644 tests/contracts/test_delivery.py create mode 100644 tests/support/receipts.py diff --git a/src/agentchat/channels/_delivery.py b/src/agentchat/channels/_delivery.py new file mode 100644 index 0000000..85dc282 --- /dev/null +++ b/src/agentchat/channels/_delivery.py @@ -0,0 +1,59 @@ +"""Bounded in-process dispatch, shared by the gateway and HTTP adapters.""" + +from __future__ import annotations + +import asyncio +import logging + +from agentchat.channels.base import Channel, MessageReceiver +from agentchat.models import Message + +logger = logging.getLogger(__name__) + + +class DeliveryTasks: + def __init__(self, limit: int, shutdown_timeout: float) -> None: + if limit < 1 or shutdown_timeout < 0: + raise ValueError("max_pending_events must be positive; shutdown_timeout nonnegative") + self.receiver: MessageReceiver | None = None + self.tasks: set[asyncio.Task[None]] = set() + self.limit = limit + self.shutdown_timeout = shutdown_timeout + self.closed = False + + def check_bound(self) -> None: + if self.receiver is None: + raise RuntimeError("Bind a receiver before running the channel") + if self.closed: + raise RuntimeError("A closed channel cannot be restarted") + + def check_reply(self) -> None: + # Accepted handlers may finish their sends while close() drains them. + if self.closed and asyncio.current_task() not in self.tasks: + raise RuntimeError("Channel is closed") + + def submit(self, channel: Channel, message: Message) -> bool: + if self.closed or len(self.tasks) >= self.limit: + return False + self.check_bound() + task = asyncio.create_task(self._deliver(channel, message)) + self.tasks.add(task) + task.add_done_callback(self.tasks.discard) + return True + + async def _deliver(self, channel: Channel, message: Message) -> None: + assert self.receiver is not None + try: + await self.receiver(channel, message) + except Exception: + logger.exception("%s message handler failed", channel.name) + + async def close(self) -> None: + self.closed = True + # A receiver may itself request shutdown. Never await/cancel the caller. + tasks = self.tasks - {asyncio.current_task()} + if tasks: + _, pending = await asyncio.wait(tasks, timeout=self.shutdown_timeout) + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) diff --git a/tests/contracts/test_delivery.py b/tests/contracts/test_delivery.py new file mode 100644 index 0000000..94f3cbf --- /dev/null +++ b/tests/contracts/test_delivery.py @@ -0,0 +1,82 @@ +import asyncio + +import pytest + +from agentchat.channels._delivery import DeliveryTasks +from tests.support.channels import FakeChannel, incoming + + +@pytest.mark.asyncio +async def test_dispatch_is_bounded_and_close_cancels_unfinished_receivers(): + delivery = DeliveryTasks(1, shutdown_timeout=0) + channel = FakeChannel() + started, cancelled = asyncio.Event(), asyncio.Event() + + async def receive(_, message): + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + delivery.receiver = receive + assert delivery.submit(channel, incoming("one", "first")) + assert not delivery.submit(channel, incoming("two", "overflow")) + await started.wait() + await delivery.close() + assert cancelled.is_set() + assert not delivery.tasks + assert not delivery.submit(channel, incoming("three", "closed")) + + +@pytest.mark.asyncio +async def test_close_drains_successful_receivers(): + delivery = DeliveryTasks(2, shutdown_timeout=1) + seen = [] + + async def receive(_, message): + seen.append(message.text) + + delivery.receiver = receive + assert delivery.submit(FakeChannel(), incoming("one", "first")) + assert delivery.submit(FakeChannel(), incoming("two", "second")) + await delivery.close() + assert seen == ["first", "second"] + assert not delivery.tasks + + +@pytest.mark.asyncio +async def test_handler_failure_is_logged_and_does_not_leak_task(caplog): + delivery = DeliveryTasks(1, shutdown_timeout=1) + + async def receive(*args): + raise RuntimeError("handler failed") + + delivery.receiver = receive + delivery.submit(FakeChannel(), incoming("one", "first")) + await delivery.close() + assert "handler failed" in caplog.text + assert not delivery.tasks + + +@pytest.mark.asyncio +async def test_receiver_can_request_shutdown_without_waiting_for_itself(): + delivery = DeliveryTasks(1, shutdown_timeout=1) + finished = asyncio.Event() + + async def receive(*args): + await delivery.close() + finished.set() + + delivery.receiver = receive + delivery.submit(FakeChannel(), incoming("one", "first")) + await asyncio.wait_for(finished.wait(), timeout=0.5) + + +def test_unbound_and_invalid_configuration(): + with pytest.raises(ValueError): + DeliveryTasks(0, 1) + with pytest.raises(ValueError): + DeliveryTasks(1, -1) + with pytest.raises(RuntimeError, match="Bind"): + DeliveryTasks(1, 1).submit(FakeChannel(), incoming("one", "first")) diff --git a/tests/support/receipts.py b/tests/support/receipts.py new file mode 100644 index 0000000..cd376fd --- /dev/null +++ b/tests/support/receipts.py @@ -0,0 +1,11 @@ +class Receipt: + def __init__(self, value="sent"): + self.value = value + self.error = None + self.calls = [] + + def get(self, *args, **kwargs): + self.calls.append((args, kwargs)) + if self.error: + raise self.error + return self.value From a60a548865bed8f2b976cc186221cda052aedc33 Mon Sep 17 00:00:00 2001 From: moe-berri Date: Mon, 12 Oct 2026 14:49:59 -0400 Subject: [PATCH 2/6] feat: add Discord gateway and single-attempt text replies --- docs/discord.md | 74 ++++++++ examples/discord_echo.py | 26 +++ pyproject.toml | 2 + src/agentchat/channels/__init__.py | 8 +- src/agentchat/channels/discord/__init__.py | 8 + src/agentchat/channels/discord/adapter.py | 133 +++++++++++++++ src/agentchat/channels/discord/events.py | 51 ++++++ src/agentchat/channels/discord/rendering.py | 7 + src/agentchat/channels/discord/transport.py | 43 +++++ tests/channels/discord/test_discord.py | 179 ++++++++++++++++++++ tests/support/discord.py | 42 +++++ 11 files changed, 572 insertions(+), 1 deletion(-) create mode 100644 docs/discord.md create mode 100644 examples/discord_echo.py create mode 100644 src/agentchat/channels/discord/__init__.py create mode 100644 src/agentchat/channels/discord/adapter.py create mode 100644 src/agentchat/channels/discord/events.py create mode 100644 src/agentchat/channels/discord/rendering.py create mode 100644 src/agentchat/channels/discord/transport.py create mode 100644 tests/channels/discord/test_discord.py create mode 100644 tests/support/discord.py diff --git a/docs/discord.md b/docs/discord.md new file mode 100644 index 0000000..46995c2 --- /dev/null +++ b/docs/discord.md @@ -0,0 +1,74 @@ +# Discord + +Connect a Discord bot through its outbound gateway. No public webhook, model API +key, or privileged Message Content intent is required for DMs and explicit mentions. + +## Get the first reply + +1. In the [Discord Developer Portal](https://discord.com/developers/applications), + create an application. Under **Bot**, create/reset its token and store it as + `DISCORD_BOT_TOKEN`. Use the bot token, not the client secret. +2. Under **OAuth2 → URL Generator**, select the `bot` scope and the permissions + **View Channels**, **Send Messages**, **Read Message History**, and **Send + Messages in Threads**. Open the generated URL and install the bot in a test + server you manage. Channel overrides must grant these permissions too. +3. From this repository, install and run: + + ```bash + python -m pip install -e '.[discord]' + export DISCORD_BOT_TOKEN='your-bot-token' + python examples/discord_echo.py + ``` + +4. Send `hello` to the bot in a DM, or `@YourBot hello` in the test channel. The + bot should reply `hello` in the same destination. Stop with Ctrl+C. + +The example echoes each accepted input. Replace its handler with your runner +after authorizing the sender. See [shared usage](providers.md) to run several +adapters or bind an existing runner directly. + +## Routing and replies + +- Each DM channel, guild channel, and native Discord thread has its own shared + conversation. Native threads set `message.thread_id` to the thread channel ID. + Use separate native threads when tasks need separate history. +- Guild messages require an explicit bot mention or a resolved reply to this bot. + When replying, leave Discord's reply mention enabled so the default gateway + intents receive the message content. Unmentioned thread messages are ignored; + this adapter does not automatically subscribe to thread follow-ups. +- Messages from bots, webhooks, and system events are ignored. Only this bot's + mention is removed from input text. Attachments become metadata references; + their contents are not fetched. +- Replies use a Discord message reference in the original channel/thread. + If the original message was deleted, the reply still posts there. Outbound + `@everyone`, role/user mentions, and reply-author pings are disabled. +- A reply must fit 2,000 UTF-16 code units. Oversized replies raise `ValueError` + before sending. Split longer output into separate `context.reply()` calls. + +`Discord(bot_token=..., max_pending_events=64, shutdown_timeout=20)` also accepts +an owned `discord.Client` through `client=` and an owned `DiscordRESTClient` +through `send_client=`. The adapter installs the client's `on_message` callback +and closes both clients on shutdown. If supplying custom intents, retain guild, +guild-message, and DM-message events. Reading unmentioned guild content requires +enabling Message Content in both the portal and your custom client; this does not +change the adapter's mention filter. + +The gateway reconnects through discord.py. Replies use a separate, single-attempt +HTTP request, including on HTTP 429 or 5xx; errors propagate. The application +owns rate-limit scheduling and recovery of uncertain sends. + +## Troubleshooting and maintenance + +| Symptom | Check | +| --- | --- | +| Login fails | Reset the bot token, update the environment, and restart. | +| Bot receives DMs but misses guild messages | Mention the bot and check channel permissions; the default does not read arbitrary channel content. | +| Replies fail with HTTP 403 | Grant Send Messages/Read Message History, plus Send Messages in Threads for a thread. | +| HTTP 429 | Reduce/pause outgoing sends and apply the platform's rate-limit policy in your host. | +| A thread gets no reply | The bot must have access; check archived/locked/private thread permissions. | + +Local tests exercise real discord.py message parsing and a local HTTP send server. +A live bot-token round trip remains the final deployment check. + +References: [discord.py gateway/intents](https://discordpy.readthedocs.io/en/stable/intents.html), +[Discord message API](https://docs.discord.com/developers/resources/message). diff --git a/examples/discord_echo.py b/examples/discord_echo.py new file mode 100644 index 0000000..df6209f --- /dev/null +++ b/examples/discord_echo.py @@ -0,0 +1,26 @@ +"""Echo messages after completing docs/discord.md; no model API key required.""" + +import asyncio +import logging + +from agentchat import AgentChat, MessageContext +from agentchat.channels.discord import Discord + + +async def main() -> None: + app = AgentChat(channels=[Discord.from_env()]) + + @app.on_message + async def respond(context: MessageContext) -> str: + # Authorize the sender here before replacing this echo with an agent. + return context.message.text or "Received your attachment." + + try: + await app.run() + finally: + await app.close() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 77d9674..5d04a98 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,8 +18,10 @@ dependencies = [ [project.optional-dependencies] openai = ["openai-agents>=0.4,<1"] +discord = ["discord.py>=2.6,<3"] dev = [ "openai-agents>=0.4,<1", + "discord.py>=2.6,<3", "pytest>=7.3,<9", "pytest-asyncio>=0.21,<1", "ruff>=0.14,<1", diff --git a/src/agentchat/channels/__init__.py b/src/agentchat/channels/__init__.py index c2b9401..26db3a0 100644 --- a/src/agentchat/channels/__init__.py +++ b/src/agentchat/channels/__init__.py @@ -1,14 +1,20 @@ +from importlib import import_module from typing import TYPE_CHECKING from agentchat.channels.base import Channel, ThreadHistoryChannel if TYPE_CHECKING: + from agentchat.channels.discord import Discord from agentchat.channels.slack import Slack, SlackUser -__all__ = ["Channel", "Slack", "SlackUser", "ThreadHistoryChannel"] +__all__ = ["Channel", "Discord", "Slack", "SlackUser", "ThreadHistoryChannel"] + +_PROVIDERS = {"Discord": "discord"} def __getattr__(name: str) -> object: + if name in _PROVIDERS: + return getattr(import_module(f"agentchat.channels.{_PROVIDERS[name]}"), name) if name in {"Slack", "SlackUser"}: try: from agentchat.channels import slack diff --git a/src/agentchat/channels/discord/__init__.py b/src/agentchat/channels/discord/__init__.py new file mode 100644 index 0000000..822e321 --- /dev/null +++ b/src/agentchat/channels/discord/__init__.py @@ -0,0 +1,8 @@ +try: + from agentchat.channels.discord.adapter import Discord +except ModuleNotFoundError as exc: + if exc.name == "discord": + raise ImportError("Install Discord support with: pip install 'agentchat[discord]'") from exc + raise + +__all__ = ["Discord"] diff --git a/src/agentchat/channels/discord/adapter.py b/src/agentchat/channels/discord/adapter.py new file mode 100644 index 0000000..1b35aa4 --- /dev/null +++ b/src/agentchat/channels/discord/adapter.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import asyncio +import logging +import os +from types import MappingProxyType + +import discord + +from agentchat.channels._delivery import DeliveryTasks +from agentchat.channels.base import MessageReceiver +from agentchat.channels.discord.events import normalize +from agentchat.channels.discord.rendering import text_payload +from agentchat.channels.discord.transport import DiscordRESTClient +from agentchat.models import Message, Sender + +logger = logging.getLogger(__name__) + + +class Discord: + """Discord gateway transport. Each DM, channel, or native thread is a conversation.""" + + name = "discord" + + def __init__( + self, + *, + bot_token: str, + client: discord.Client | None = None, + send_client: DiscordRESTClient | None = None, + max_pending_events: int = 64, + shutdown_timeout: float = 20, + ) -> None: + if not bot_token.strip(): + raise ValueError("Discord bot_token is required") + self._token = bot_token + self._delivery = DeliveryTasks(max_pending_events, shutdown_timeout) + intents = discord.Intents.none() + intents.guilds = True + intents.guild_messages = True + intents.dm_messages = True + self._client = client or discord.Client( + intents=intents, + allowed_mentions=discord.AllowedMentions.none(), + ) + # This adapter owns the injected client's message callback and lifecycle. + self._client.on_message = self._on_message + self._sender = send_client or DiscordRESTClient(bot_token) + self._close_lock = asyncio.Lock() + self._closed = False + + @classmethod + def from_env(cls) -> Discord: + token = os.getenv("DISCORD_BOT_TOKEN") + if not token: + raise RuntimeError("DISCORD_BOT_TOKEN is required") + return cls(bot_token=token) + + def bind(self, receiver: MessageReceiver) -> None: + self._delivery.receiver = receiver + + async def run(self) -> None: + self._delivery.check_bound() + try: + await self._client.start(self._token) + finally: + await self.close() + + async def close(self) -> None: + async with self._close_lock: + if self._closed: + return + self._closed = True + # Stop admitting events, but retain the HTTP client while replies drain. + self._delivery.closed = True + try: + await self._delivery.close() + finally: + try: + await self._client.close() + finally: + await self._sender.close() + + async def _on_message(self, event: discord.Message) -> None: + if self._client.user is None or self._delivery.closed: + return + message = normalize(event, bot_id=self._client.user.id) + if message is not None and not self._delivery.submit(self, message): + logger.warning("Discord event dropped: max_pending_events reached") + + async def reply(self, source: Message, content: str) -> Message: + content = text_payload(content) + bot = self._client.user + data = source.metadata + channel_id, message_id = data.get("channel_id"), data.get("message_id") + if ( + source.channel != self.name + or bot is None + or data.get("bot_id") != bot.id + or not isinstance(channel_id, int) + or not isinstance(message_id, int) + ): + raise ValueError("Source does not belong to this Discord adapter") + if self._client.is_closed(): + raise RuntimeError("Discord connection is closed") + self._delivery.check_reply() + reference = { + "message_id": str(message_id), + "channel_id": str(channel_id), + "fail_if_not_exists": False, + } + if data.get("guild_id") is not None: + reference["guild_id"] = str(data["guild_id"]) + sent_id = await self._sender.send_text( + channel_id, + { + "content": content, + "message_reference": reference, + "allowed_mentions": {"parse": [], "replied_user": False}, + }, + ) + if not isinstance(sent_id, int) or sent_id <= 0: + raise RuntimeError("Discord did not return a message receipt") + return Message( + id=f"discord:{bot.id}:{channel_id}:{sent_id}", + conversation_id=source.conversation_id, + channel=self.name, + sender=Sender(id=str(bot.id), display_name=bot.display_name), + text=content, + role="assistant", + thread_id=source.thread_id, + metadata=MappingProxyType({**data, "message_id": sent_id}), + ) diff --git a/src/agentchat/channels/discord/events.py b/src/agentchat/channels/discord/events.py new file mode 100644 index 0000000..5cd0b02 --- /dev/null +++ b/src/agentchat/channels/discord/events.py @@ -0,0 +1,51 @@ +"""Discord gateway messages to SDK values; no network calls.""" + +from __future__ import annotations + +import re +from types import MappingProxyType + +import discord + +from agentchat.models import Attachment, Message, Sender + + +def normalize(event: discord.Message, *, bot_id: int) -> Message | None: + if event.author.bot or event.webhook_id or event.author.id == bot_id or event.is_system(): + return None + direct = event.guild is None + mentioned = any(user.id == bot_id for user in event.mentions) + reference = event.reference.resolved if event.reference else None + replies_to_bot = isinstance(reference, discord.Message) and reference.author.id == bot_id + if not (direct or mentioned or replies_to_bot): + return None + text = re.sub(rf"<@!?{bot_id}>", "", event.content).strip() + attachments = tuple( + Attachment( + id=str(item.id), name=item.filename, media_type=item.content_type, size=item.size + ) + for item in event.attachments + ) + if not text and not attachments: + return None + channel_id = event.channel.id + guild_id = event.guild.id if event.guild else None + thread_id = str(channel_id) if isinstance(event.channel, discord.Thread) else None + return Message( + id=f"discord:{bot_id}:{channel_id}:{event.id}", + conversation_id=f"discord:{bot_id}:{guild_id or 'dm'}:{channel_id}", + channel="discord", + sender=Sender(id=str(event.author.id), display_name=event.author.display_name), + text=text, + role="user", + thread_id=thread_id, + attachments=attachments, + metadata=MappingProxyType( + { + "bot_id": bot_id, + "channel_id": channel_id, + "guild_id": guild_id, + "message_id": event.id, + } + ), + ) diff --git a/src/agentchat/channels/discord/rendering.py b/src/agentchat/channels/discord/rendering.py new file mode 100644 index 0000000..f04e2c5 --- /dev/null +++ b/src/agentchat/channels/discord/rendering.py @@ -0,0 +1,7 @@ +def text_payload(content: str) -> str: + if not isinstance(content, str) or not content.strip(): + raise ValueError("Discord replies require nonempty text") + # Count UTF-16 units as well, so astral characters cannot exceed the platform limit. + if len(content.encode("utf-16-le")) // 2 > 2000: + raise ValueError("Discord replies must fit 2,000 characters; send separate replies") + return content diff --git a/src/agentchat/channels/discord/transport.py b/src/agentchat/channels/discord/transport.py new file mode 100644 index 0000000..39b3d52 --- /dev/null +++ b/src/agentchat/channels/discord/transport.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import aiohttp + +API_ROOT = "https://discord.com/api/v10" + + +class DiscordRESTClient: + """One-attempt message sends; discord.py's gateway owns reconnects separately.""" + + def __init__(self, bot_token: str) -> None: + self._token = bot_token + self._session: aiohttp.ClientSession | None = None + self._closed = False + + async def send_text(self, channel_id: int, payload: dict) -> int: + if self._closed: + raise RuntimeError("Discord send client is closed") + if self._session is None: + self._session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) + try: + async with self._session.post( + f"{API_ROOT}/channels/{channel_id}/messages", + headers={"Authorization": f"Bot {self._token}"}, + json=payload, + allow_redirects=False, + ) as response: + if not 200 <= response.status < 300: + raise RuntimeError(f"Discord send failed (HTTP {response.status})") + result = await response.json() + except (aiohttp.ClientError, TimeoutError, ValueError): + raise RuntimeError("Discord request failed; delivery may be uncertain") from None + message_id = result.get("id") if isinstance(result, dict) else None + if not isinstance(message_id, str) or not message_id.isascii() or not message_id.isdigit(): + raise RuntimeError("Discord did not return a message receipt") + if int(message_id) <= 0: + raise RuntimeError("Discord did not return a message receipt") + return int(message_id) + + async def close(self) -> None: + self._closed = True + if self._session: + await self._session.close() diff --git a/tests/channels/discord/test_discord.py b/tests/channels/discord/test_discord.py new file mode 100644 index 0000000..c847a68 --- /dev/null +++ b/tests/channels/discord/test_discord.py @@ -0,0 +1,179 @@ +from types import SimpleNamespace + +import discord +import pytest +from aiohttp import web +from aiohttp.test_utils import TestServer + +from agentchat.channels.discord import Discord +from agentchat.channels.discord.events import normalize +from agentchat.channels.discord.rendering import text_payload +from agentchat.channels.discord.transport import DiscordRESTClient +from tests.support.discord import DiscordClient, discord_event +from tests.support.receipts import Receipt + + +def test_dm_normalizes_a_real_discord_py_message(): + client = discord.Client(intents=discord.Intents.none()) + state = client._connection + state.user = discord.ClientUser( + state=state, + data={ + "id": "99", + "username": "Bot", + "discriminator": "0", + "bot": True, + "avatar": None, + }, + ) + author = {"id": "30", "username": "Alice", "discriminator": "0", "avatar": None} + channel = discord.DMChannel( + me=state.user, state=state, data={"id": "10", "recipients": [author]} + ) + event = discord.Message( + state=state, + channel=channel, + data={ + "id": "100", + "type": 0, + "content": "hello", + "author": author, + "attachments": [], + "embeds": [], + "edited_timestamp": None, + "pinned": False, + "mention_everyone": False, + "tts": False, + "mentions": [], + }, + ) + message = normalize(event, bot_id=99) + assert message.id == "discord:99:10:100" + assert message.conversation_id == "discord:99:dm:10" + assert message.sender.display_name == "Alice" + assert message.thread_id is None + + +def test_mentions_and_threads_are_scoped_to_provider_channels(): + event = discord_event( + direct=False, + thread=True, + content="<@99> hi <@!99> <@42>", + mentions=[SimpleNamespace(id=99)], + ) + message = normalize(event, bot_id=99) + assert message.text == "hi <@42>" + assert message.thread_id == "10" + assert message.conversation_id == "discord:99:20:10" + event.channel.id = 11 + assert normalize(event, bot_id=99).conversation_id != message.conversation_id + + +def test_reply_to_bot_and_attachment_only_dm(): + reference = discord_event(author=SimpleNamespace(id=99, bot=True)) + event = discord_event(direct=False, reference=SimpleNamespace(resolved=reference)) + assert normalize(event, bot_id=99) is not None + event = discord_event( + content="", + attachments=[ + SimpleNamespace( + id=7, + filename="picture.png", + content_type="image/png", + size=12, + url="https://example.com/never-retained", + ) + ], + ) + message = normalize(event, bot_id=99) + assert message.attachments[0].id == "7" + assert "never-retained" not in repr(message) + + +@pytest.mark.parametrize( + "changes", + [ + {"direct": False}, + {"content": ""}, + {"webhook_id": 9}, + {"author": SimpleNamespace(id=99, bot=False)}, + {"author": SimpleNamespace(id=77, bot=True)}, + ], +) +def test_ignores_unaddressed_bot_webhook_and_empty_events(changes): + assert normalize(discord_event(**changes), bot_id=99) is None + + +@pytest.mark.asyncio +async def test_reply_disables_mentions_and_uses_message_reference(): + receipt = Receipt(123) + client = DiscordClient(receipt) + channel = Discord(bot_token="fake", client=client, send_client=client) + source = normalize(discord_event(), bot_id=99) + try: + await channel.reply(source, "@everyone <@30>") + destination, options = receipt.calls[0][0] + assert destination == 10 + assert options["allowed_mentions"] == {"parse": [], "replied_user": False} + assert options["message_reference"]["message_id"] == "100" + assert options["message_reference"]["fail_if_not_exists"] is False + finally: + await channel.close() + + +def test_limits_reject_before_network(): + assert text_payload("a" * 2000) == "a" * 2000 + with pytest.raises(ValueError, match="2,000"): + text_payload("a" * 2001) + with pytest.raises(ValueError, match="2,000"): + text_payload("😀" * 1001) + + +@pytest.mark.asyncio +async def test_login_failure_closes_client(): + client = DiscordClient(Receipt()) + + async def fail(token): + raise discord.LoginFailure("invalid token") + + client.start = fail + channel = Discord(bot_token="fake", client=client) + + async def receive(*args): + pass + + channel.bind(receive) + with pytest.raises(discord.LoginFailure): + await channel.run() + assert client.closed + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [200, 429, 500, 302]) +async def test_http_send_receipt_and_no_retries(monkeypatch, status): + calls = [] + + async def send(request): + assert request.headers["Authorization"] == "Bot fake" + calls.append(await request.json()) + return web.json_response({"id": "123"}, status=status, headers={"Location": "/no-redirect"}) + + app = web.Application() + app.router.add_post("/channels/10/messages", send) + async with TestServer(app) as server: + monkeypatch.setattr( + "agentchat.channels.discord.transport.API_ROOT", + str( + server.make_url("/"), + ).rstrip("/"), + ) + client = DiscordRESTClient("fake") + try: + if status == 200: + assert await client.send_text(10, {"content": "hello"}) == 123 + else: + with pytest.raises(RuntimeError, match=f"HTTP {status}"): + await client.send_text(10, {"content": "hello"}) + assert len(calls) == 1 + finally: + await client.close() diff --git a/tests/support/discord.py b/tests/support/discord.py new file mode 100644 index 0000000..6b10159 --- /dev/null +++ b/tests/support/discord.py @@ -0,0 +1,42 @@ +import asyncio +from types import SimpleNamespace +from unittest.mock import Mock + +import discord + + +def discord_event(*, direct=True, thread=False, **changes): + channel = Mock(spec=discord.Thread) if thread else SimpleNamespace() + channel.id = 10 + event = Mock(spec=discord.Message) + event.id, event.channel = 100, channel + event.guild = None if direct else SimpleNamespace(id=20) + event.author = SimpleNamespace(id=30, bot=False, display_name="Alice") + event.webhook_id, event.reference = None, None + event.content, event.mentions, event.attachments = "hello", [], [] + event.is_system.return_value = False + for key, value in changes.items(): + setattr(event, key, value) + return event + + +class DiscordClient: + def __init__(self, receipt): + self.receipt = receipt + self.user = SimpleNamespace(id=99, display_name="Bot") + self.stopped, self.started = asyncio.Event(), asyncio.Event() + self.closed = False + + async def start(self, token): + self.started.set() + await self.stopped.wait() + + async def close(self): + self.closed = True + self.stopped.set() + + def is_closed(self): + return self.closed + + async def send_text(self, channel_id, payload): + return self.receipt.get(channel_id, payload) From c5b424486d594e0ec5051e0bbe2c656bf887dca1 Mon Sep 17 00:00:00 2001 From: moe-berri Date: Mon, 12 Oct 2026 14:50:18 -0400 Subject: [PATCH 3/6] feat: add authenticated Microsoft Teams conversations --- docs/teams.md | 84 +++++++++ examples/teams_echo.py | 26 +++ pyproject.toml | 6 + src/agentchat/channels/__init__.py | 5 +- src/agentchat/channels/teams/__init__.py | 8 + src/agentchat/channels/teams/adapter.py | 201 +++++++++++++++++++++ src/agentchat/channels/teams/events.py | 118 +++++++++++++ src/agentchat/channels/teams/rendering.py | 7 + tests/channels/teams/test_teams.py | 204 ++++++++++++++++++++++ tests/support/teams.py | 48 +++++ 10 files changed, 705 insertions(+), 2 deletions(-) create mode 100644 docs/teams.md create mode 100644 examples/teams_echo.py create mode 100644 src/agentchat/channels/teams/__init__.py create mode 100644 src/agentchat/channels/teams/adapter.py create mode 100644 src/agentchat/channels/teams/events.py create mode 100644 src/agentchat/channels/teams/rendering.py create mode 100644 tests/channels/teams/test_teams.py create mode 100644 tests/support/teams.py diff --git a/docs/teams.md b/docs/teams.md new file mode 100644 index 0000000..c618560 --- /dev/null +++ b/docs/teams.md @@ -0,0 +1,84 @@ +# Microsoft Teams + +Connect a single-tenant Teams bot in Microsoft's public cloud using the +Microsoft 365 Agents SDK. You need a tenant that permits installing your Teams +app, an Azure Bot registration, and a public HTTPS endpoint. A tenant administrator +may need to approve custom app upload or installation. + +## Get the first reply + +1. Create a **single-tenant** Microsoft Entra application and a client secret. + Record its **Application (client) ID**, **Directory (tenant) ID**, and the + secret **value** (not the secret ID). Create an Azure Bot resource using this + app ID and tenant; enable its **Microsoft Teams** channel. Microsoft's + [Agents SDK quickstart](https://learn.microsoft.com/en-us/microsoft-365/agents-sdk/quickstart) + describes the registration options. +2. From this repository, install and start the echo server: + + ```bash + python -m pip install -e '.[teams]' + export TEAMS_APP_ID='application-client-id' + export TEAMS_TENANT_ID='directory-tenant-id' + export TEAMS_APP_PASSWORD='client-secret-value' + python examples/teams_echo.py + ``` + + It listens on `127.0.0.1:3978`. Forward a public HTTPS URL through a development + tunnel to this port, or deploy behind an HTTPS reverse proxy. For containers, + set `TEAMS_HOST=0.0.0.0`; `TEAMS_PORT` overrides the port. +3. Set the Azure Bot **Messaging endpoint** to + `https://YOUR_HOST/api/messages`. Keep bearer authorization enabled. Anonymous + curl requests are expected to return 401; they do not test a Teams conversation. +4. In the [Teams Developer Portal](https://dev.teams.microsoft.com/), create a + Teams app and add a bot capability using the same app ID. Enable **Personal** + scope first; optionally add **Team** and **Group Chat** scopes. Complete the + portal's required app metadata/icons, download the app package, and upload it + through Teams **Apps → Manage your apps → Upload an app** (subject to tenant + policy). Install it for yourself or in your test team. +5. Send `hello` in the bot's personal chat. Expect `hello` back. In a team/channel + or group chat, send `@YourBot hello`. Stop the local server with Ctrl+C. + +## Routing and hosting + +Personal chats accept ordinary messages. Group chats and channels require a bot +mention on each incoming message. Personal and group chats retain their shared +conversation; channel replies preserve the original root thread. Only the bot's +own mention is removed. Sender IDs remain Teams IDs scoped by the configured +tenant; display names do not authorize access. + +`Teams(app_id=..., app_password=..., tenant_id=...)` validates Microsoft JWTs, +audience/issuer, the configured tenant, and HTTPS Microsoft service URLs. This +adapter supports public-cloud single-tenant client-secret authentication. It does +not add Graph chat-reading permissions, tenant-wide message subscriptions, +anonymous emulator access, or sovereign-cloud configuration. + +For an existing aiohttp host, use `teams.create_app()` and manage its runner +yourself. This keeps the authentication middleware and `/api/messages` route. +Bind a receiver before serving requests and call `await teams.close()` at +shutdown. Use either your host or `teams.run()`, not both. Supplying `adapter=` +injects a Microsoft `CloudAdapter` for tests or a host-owned connector factory; +the adapter's turn-error handler is replaced so errors propagate rather than +posting diagnostics to the conversation. + +Normal messages are acknowledged with HTTP 202 after admission to bounded +background work; replies use a fresh continuation context, not an expired HTTP +turn. A full queue returns 503 before admission. See [delivery semantics](providers.md#delivery-and-shutdown). + +Replies use plain text and accept up to 24,000 UTF-8 bytes, leaving room for the +activity envelope below Teams' message limit. Split larger answers into separate +replies. Incoming file/card metadata is retained, but download, upload, card +actions, rich replies, and native-history capabilities are not implemented here. + +## Troubleshooting and maintenance + +| Symptom | Check | +| --- | --- | +| HTTP 401 from real Teams traffic | Azure Bot app ID, tenant, client-secret value/expiry, and proxy forwarding of `Authorization`. | +| Bot not available in Teams | Teams channel enabled in Azure Bot; matching manifest bot ID; tenant custom-app policy. | +| DM works but channel does not | Install the app in that team and mention it on each message. | +| No traffic at `/api/messages` | Public HTTPS endpoint and tunnel/proxy path point to this process. | +| Send fails after setup worked | Rotate an expired client secret and restart; verify the app remains installed. | + +Local tests use Microsoft's real activity, continuation, and routing code with +fake connectors, plus unauthenticated-request rejection. They do not validate +your Azure registration. The first live Teams reply above verifies that setup. diff --git a/examples/teams_echo.py b/examples/teams_echo.py new file mode 100644 index 0000000..3dc6041 --- /dev/null +++ b/examples/teams_echo.py @@ -0,0 +1,26 @@ +"""Echo messages after completing docs/teams.md; no model API key required.""" + +import asyncio +import logging + +from agentchat import AgentChat, MessageContext +from agentchat.channels.teams import Teams + + +async def main() -> None: + app = AgentChat(channels=[Teams.from_env()]) + + @app.on_message + async def respond(context: MessageContext) -> str: + # Authorize the sender here before replacing this echo with an agent. + return context.message.text or "Received your attachment." + + try: + await app.run() + finally: + await app.close() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 5d04a98..ea66686 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,9 +19,15 @@ dependencies = [ [project.optional-dependencies] openai = ["openai-agents>=0.4,<1"] discord = ["discord.py>=2.6,<3"] +teams = [ + "microsoft-agents-hosting-aiohttp>=1.8,<2", + "microsoft-agents-authentication-msal>=1.8,<2", +] dev = [ "openai-agents>=0.4,<1", "discord.py>=2.6,<3", + "microsoft-agents-hosting-aiohttp>=1.8,<2", + "microsoft-agents-authentication-msal>=1.8,<2", "pytest>=7.3,<9", "pytest-asyncio>=0.21,<1", "ruff>=0.14,<1", diff --git a/src/agentchat/channels/__init__.py b/src/agentchat/channels/__init__.py index 26db3a0..a0bf71d 100644 --- a/src/agentchat/channels/__init__.py +++ b/src/agentchat/channels/__init__.py @@ -6,10 +6,11 @@ if TYPE_CHECKING: from agentchat.channels.discord import Discord from agentchat.channels.slack import Slack, SlackUser + from agentchat.channels.teams import Teams -__all__ = ["Channel", "Discord", "Slack", "SlackUser", "ThreadHistoryChannel"] +__all__ = ["Channel", "Discord", "Slack", "SlackUser", "Teams", "ThreadHistoryChannel"] -_PROVIDERS = {"Discord": "discord"} +_PROVIDERS = {"Discord": "discord", "Teams": "teams"} def __getattr__(name: str) -> object: diff --git a/src/agentchat/channels/teams/__init__.py b/src/agentchat/channels/teams/__init__.py new file mode 100644 index 0000000..9d0f201 --- /dev/null +++ b/src/agentchat/channels/teams/__init__.py @@ -0,0 +1,8 @@ +try: + from agentchat.channels.teams.adapter import Teams +except ModuleNotFoundError as exc: + if exc.name and exc.name.startswith("microsoft_agents"): + raise ImportError("Install Teams support with: pip install 'agentchat[teams]'") from exc + raise + +__all__ = ["Teams"] diff --git a/src/agentchat/channels/teams/adapter.py b/src/agentchat/channels/teams/adapter.py new file mode 100644 index 0000000..b4f40a9 --- /dev/null +++ b/src/agentchat/channels/teams/adapter.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +import asyncio +import os +from types import MappingProxyType +from urllib.parse import quote, urlsplit + +from aiohttp import web +from microsoft_agents.activity import Activity +from microsoft_agents.authentication.msal import MsalConnectionManager +from microsoft_agents.hosting.aiohttp import CloudAdapter, jwt_authorization_middleware +from microsoft_agents.hosting.core import AgentAuthConfiguration, OutboundHostValidator, TurnContext +from pydantic import ValidationError + +from agentchat.channels._delivery import DeliveryTasks +from agentchat.channels.base import MessageReceiver +from agentchat.channels.teams.events import normalize +from agentchat.channels.teams.rendering import text_payload +from agentchat.models import Message, Sender + + +class Teams: + """Single-tenant Teams bot using the Microsoft 365 Agents SDK (public cloud).""" + + name = "teams" + + def __init__( + self, + *, + app_id: str, + app_password: str, + tenant_id: str, + host: str = "127.0.0.1", + port: int = 3978, + adapter: CloudAdapter | None = None, + max_pending_events: int = 64, + shutdown_timeout: float = 20, + ) -> None: + if not all(value.strip() for value in (app_id, app_password, tenant_id)): + raise ValueError("Teams app_id, app_password, and tenant_id are required") + self._app_id, self._tenant_id = app_id, tenant_id + self._host, self._port = host, port + self._delivery = DeliveryTasks(max_pending_events, shutdown_timeout) + self._auth = AgentAuthConfiguration( + client_id=app_id, + client_secret=app_password, + tenant_id=tenant_id, + anonymous_allowed=False, + validate_issuer=True, + ) + self._hosts = OutboundHostValidator(enabled=True) + self._adapter = adapter or CloudAdapter( + connection_manager=MsalConnectionManager( + connections_configurations={"SERVICE_CONNECTION": self._auth}, + ), + host_validator=self._hosts, + ) + self._adapter.on_turn_error = self._turn_error + self._runner: web.AppRunner | None = None + self._stopped = asyncio.Event() + self._close_lock = asyncio.Lock() + self._closed = False + + @classmethod + def from_env(cls) -> Teams: + keys = ("TEAMS_APP_ID", "TEAMS_APP_PASSWORD", "TEAMS_TENANT_ID") + values = [os.getenv(key) for key in keys] + if not all(values): + raise RuntimeError(f"{', '.join(keys)} are required") + return cls( + app_id=values[0], + app_password=values[1], + tenant_id=values[2], + host=os.getenv("TEAMS_HOST", "127.0.0.1"), + port=int(os.getenv("TEAMS_PORT", "3978")), + ) + + def bind(self, receiver: MessageReceiver) -> None: + self._delivery.receiver = receiver + + def create_app(self) -> web.Application: + """Host this app yourself, or use run(). Authentication is always installed.""" + app = web.Application(middlewares=[jwt_authorization_middleware]) + # These keys are part of Microsoft's aiohttp middleware contract. + app["agent_configuration"] = self._auth + app.router.add_post("/api/messages", self._handle_request) + return app + + async def run(self) -> None: + self._delivery.check_bound() + self._runner = web.AppRunner(self.create_app(), access_log=None) + try: + await self._runner.setup() + await web.TCPSite(self._runner, self._host, self._port).start() + await self._stopped.wait() + finally: + await self.close() + + async def close(self) -> None: + async with self._close_lock: + if self._closed: + return + self._closed = True + self._delivery.closed = True + self._stopped.set() + try: + if self._runner: + await self._runner.cleanup() + finally: + await self._delivery.close() + + async def _handle_request(self, request: web.Request) -> web.Response: + if self._delivery.closed: + raise web.HTTPServiceUnavailable() + try: + event = await request.json() + if not isinstance(event, dict) or not isinstance(event.get("serviceUrl"), str): + raise ValueError("Missing Teams service URL") + # Validate before CloudAdapter allocates any token-bearing connector clients. + self._validate_service_url(event["serviceUrl"]) + return await self._adapter.process(request, self) + except (ValidationError, ValueError, TypeError): + raise web.HTTPBadRequest(text="Invalid Teams activity") from None + + @staticmethod + async def _turn_error(context: TurnContext, error: Exception) -> None: + # Microsoft's default sends exception details to the conversation. + raise error + + async def on_turn(self, context: TurnContext) -> None: + """Called by the authenticated Microsoft adapter, never directly by ingress.""" + event = context.activity.model_dump(by_alias=True, exclude_none=True) + message = normalize(event, app_id=self._app_id, tenant_id=self._tenant_id) + if message is None: + return + self._validate_service_url(event["serviceUrl"]) + if not self._delivery.submit(self, message): + raise web.HTTPServiceUnavailable(text="Teams receiver is busy") + + def _validate_service_url(self, url: str) -> None: + parts = urlsplit(url) + if ( + parts.scheme != "https" + or parts.username + or parts.password + or parts.query + or parts.fragment + or parts.port not in (None, 443) + or not self._hosts.is_allowed(url) + ): + raise ValueError("Teams service URL is not an allowed HTTPS Microsoft endpoint") + + async def reply(self, source: Message, content: str) -> Message: + content = text_payload(content) + data = source.metadata + route = data.get("continuation_activity") + if ( + source.channel != self.name + or data.get("app_id") != self._app_id + or data.get("tenant_id") != self._tenant_id + or not isinstance(route, dict) + ): + raise ValueError("Source does not belong to this Teams adapter") + activity = Activity.model_validate(route) + if ( + not activity.conversation + or not activity.conversation.id + or not activity.recipient + or activity.recipient.id not in (self._app_id, f"28:{self._app_id}") + or route.get("channelData", {}).get("tenant", {}).get("id") != self._tenant_id + ): + raise ValueError("Invalid Teams conversation reference") + self._validate_service_url(activity.service_url) + self._delivery.check_reply() + receipt_id = None + + async def send(context: TurnContext) -> None: + nonlocal receipt_id + receipt = await context.send_activity( + Activity( + type="message", + text=content, + text_format="plain", + ) + ) + receipt_id = receipt.id if receipt else None + + await self._adapter.continue_conversation(self._app_id, activity, send) + if not isinstance(receipt_id, str) or not receipt_id: + raise RuntimeError("Teams did not return a message receipt") + return Message( + id=f"teams:{self._app_id}:{self._tenant_id}:" + f"{quote(activity.conversation.id, safe='')}:{receipt_id}", + conversation_id=source.conversation_id, + channel=self.name, + sender=Sender(id=activity.recipient.id, display_name=activity.recipient.name), + text=content, + role="assistant", + thread_id=source.thread_id, + metadata=MappingProxyType({**data, "activity_id": receipt_id}), + ) diff --git a/src/agentchat/channels/teams/events.py b/src/agentchat/channels/teams/events.py new file mode 100644 index 0000000..538297c --- /dev/null +++ b/src/agentchat/channels/teams/events.py @@ -0,0 +1,118 @@ +"""Normalize authenticated Teams activities without retaining request/turn objects.""" + +from __future__ import annotations + +from collections.abc import Mapping +from types import MappingProxyType +from urllib.parse import quote + +from agentchat.models import Attachment, Message, Sender + + +def _items(value: object) -> list: + return value if isinstance(value, list) else [] + + +def normalize(event: Mapping[str, object], *, app_id: str, tenant_id: str) -> Message | None: + if event.get("type") != "message" or event.get("channelId") != "msteams": + return None + conversation, sender = event.get("conversation"), event.get("from") + recipient, data = event.get("recipient"), event.get("channelData") + if not all(isinstance(value, dict) for value in (conversation, sender, recipient, data)): + return None + tenant_data = data.get("tenant") + tenant = tenant_data.get("id") if isinstance(tenant_data, dict) else None + if tenant != tenant_id or recipient.get("id") not in (app_id, f"28:{app_id}"): + return None + activity_id, sender_id, conversation_id = ( + event.get("id"), + sender.get("id"), + conversation.get("id"), + ) + if not all( + isinstance(value, str) and value + for value in ( + activity_id, + sender_id, + conversation_id, + event.get("serviceUrl"), + ) + ): + return None + if sender_id == recipient.get("id") or sender.get("role") == "bot": + return None + kind = conversation.get("conversationType") + if kind not in ("personal", "groupChat", "channel"): + return None + text = event.get("text") or "" + if not isinstance(text, str): + return None + mentioned = False + for entity in _items(event.get("entities")): + if ( + isinstance(entity, dict) + and entity.get("type") == "mention" + and isinstance(entity.get("mentioned"), dict) + and entity["mentioned"].get("id") == recipient["id"] + ): + mentioned = True + if isinstance(entity.get("text"), str): + text = text.replace(entity["text"], "") + if kind != "personal" and not mentioned: + return None + attachments = [] + for index, item in enumerate(_items(event.get("attachments"))): + if not isinstance(item, dict): + continue + # Teams uses uniqueId for download-info files; never expose contentUrl as an ID. + info = item.get("content") + file_id = info.get("uniqueId") if isinstance(info, dict) else None + attachments.append( + Attachment( + id=file_id if isinstance(file_id, str) and file_id else f"{activity_id}:{index}", + name=item.get("name"), + media_type=item.get("contentType"), + ) + ) + if not text.strip() and not attachments: + return None + root = None + if kind == "channel": + suffix = conversation_id.partition(";messageid=")[2] + root = event.get("replyToId") or suffix or activity_id + if not isinstance(root, str): + return None + # Teams may append ;messageid= to channel conversation IDs. + base_id = conversation_id.split(";messageid=", 1)[0] if root else conversation_id + identity = ":".join(quote(value, safe="") for value in (app_id, tenant_id, base_id)) + if root: + identity += f":{quote(root, safe='')}" + route = { + "type": "message", + "channelId": "msteams", + "serviceUrl": event["serviceUrl"], + "conversation": dict(conversation), + "from": dict(sender), + "recipient": dict(recipient), + "channelData": {"tenant": {"id": tenant_id}}, + } + if root: + route["id"] = root + return Message( + id=f"teams:{app_id}:{tenant_id}:{quote(conversation_id, safe='')}:{activity_id}", + conversation_id=f"teams:{identity}", + channel="teams", + sender=Sender(id=sender_id, display_name=sender.get("name")), + text=text.strip(), + role="user", + thread_id=root, + attachments=tuple(attachments), + metadata=MappingProxyType( + { + "app_id": app_id, + "tenant_id": tenant_id, + "activity_id": activity_id, + "continuation_activity": route, + } + ), + ) diff --git a/src/agentchat/channels/teams/rendering.py b/src/agentchat/channels/teams/rendering.py new file mode 100644 index 0000000..81fba17 --- /dev/null +++ b/src/agentchat/channels/teams/rendering.py @@ -0,0 +1,7 @@ +def text_payload(content: str) -> str: + if not isinstance(content, str) or not content.strip(): + raise ValueError("Teams replies require nonempty text") + # Leave space for the activity envelope below Teams' 40 KB message limit. + if len(content.encode("utf-8")) > 24_000: + raise ValueError("Teams text replies must fit 24,000 UTF-8 bytes; send separate replies") + return content diff --git a/tests/channels/teams/test_teams.py b/tests/channels/teams/test_teams.py new file mode 100644 index 0000000..8756b84 --- /dev/null +++ b/tests/channels/teams/test_teams.py @@ -0,0 +1,204 @@ +import asyncio +import importlib +from dataclasses import replace +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from aiohttp.test_utils import TestClient, TestServer +from microsoft_agents.activity import Activity +from microsoft_agents.hosting.core import ClaimsIdentity + +from agentchat.channels.teams import Teams +from agentchat.channels.teams.events import normalize +from tests.support.receipts import Receipt +from tests.support.teams import APP_ID, TENANT_ID, teams_adapter, teams_event + + +def make_channel(**options): + receipt = Receipt() + adapter, connector, factory = teams_adapter(receipt) + channel = Teams( + app_id=APP_ID, + tenant_id=TENANT_ID, + app_password="fake", + adapter=adapter, + **options, + ) + return channel, receipt, connector, factory + + +def mention(): + return {"type": "mention", "text": "Bot", "mentioned": {"id": f"28:{APP_ID}"}} + + +@pytest.mark.parametrize("kind", ["personal", "groupChat", "channel"]) +@pytest.mark.asyncio +async def test_actual_microsoft_sdk_reply_routing(kind): + channel, receipt, connector, _ = make_channel() + event = teams_event(kind=kind, text="Bot hello", entities=[mention()]) + if kind == "channel": + event["replyToId"] = "root" + source = normalize(event, app_id=APP_ID, tenant_id=TENANT_ID) + try: + response = await channel.reply(source, "answer") + args, _ = receipt.calls[0] + sent = args[-1] + assert sent.conversation.id == "conversation" + assert sent.from_property.id == f"28:{APP_ID}" + assert sent.recipient.id == "29:alice" + assert sent.reply_to_id == ("root" if kind == "channel" else None) + assert sent.text_format == "plain" + assert response.thread_id == source.thread_id + connector.close.assert_awaited_once() + finally: + await channel.close() + + +def test_channel_roots_and_mentions_preserve_identity(): + first = teams_event(kind="channel", id="root", entities=[mention()]) + reply = teams_event(kind="channel", id="reply", replyToId="root", entities=[mention()]) + reply["conversation"]["id"] += ";messageid=root" + one = normalize(first, app_id=APP_ID, tenant_id=TENANT_ID) + two = normalize(reply, app_id=APP_ID, tenant_id=TENANT_ID) + assert one.conversation_id == two.conversation_id + assert one.id != two.id + assert one.thread_id == two.thread_id == "root" + reply["replyToId"] = "other-root" + assert ( + normalize(reply, app_id=APP_ID, tenant_id=TENANT_ID).conversation_id != one.conversation_id + ) + + +@pytest.mark.parametrize( + "changes", + [ + {"type": "typing"}, + {"channelId": "webchat"}, + {"kind": "channel"}, + {"channelData": {"tenant": {"id": "other"}}}, + {"channelData": {"tenant": None}}, + {"recipient": {"id": "other-bot"}}, + {"from": {"id": f"28:{APP_ID}"}}, + {"conversation": {"id": "c", "conversationType": "unknown"}}, + {"text": ""}, + ], +) +def test_ignores_nonmessages_wrong_tenant_and_unaddressed_group_events(changes): + assert normalize(teams_event(**changes), app_id=APP_ID, tenant_id=TENANT_ID) is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("authorization", [None, "Basic x", "Bearer invalid"]) +async def test_endpoint_rejects_unauthenticated_requests(authorization): + channel, receipt, _, _ = make_channel() + async with TestClient(TestServer(channel.create_app())) as client: + headers = {"Authorization": authorization} if authorization else {} + response = await client.post("/api/messages", json=teams_event(), headers=headers) + assert response.status == 401 + assert not receipt.calls + await channel.close() + + +@pytest.mark.asyncio +async def test_authenticated_endpoint_returns_before_handler_and_backpressures(monkeypatch): + channel, _, _, _ = make_channel(max_pending_events=1, shutdown_timeout=0) + started = asyncio.Event() + + async def receive(*args): + started.set() + await asyncio.Event().wait() + + channel.bind(receive) + # Stub the validated identity at the JWT boundary; exercise the real CloudAdapter below it. + middleware = importlib.import_module( + "microsoft_agents.hosting.aiohttp.jwt_authorization_middleware", + ) + monkeypatch.setattr( + middleware, + "_authorize_request", + AsyncMock( + return_value=ClaimsIdentity( + {"aud": APP_ID, "iss": "https://api.botframework.com"}, + "JWT", + ) + ), + ) + try: + async with TestClient(TestServer(channel.create_app())) as client: + response = await client.post("/api/messages", json=teams_event()) + assert response.status == 202 + await asyncio.wait_for(started.wait(), timeout=1) + response = await client.post("/api/messages", json=teams_event(id="second")) + assert response.status == 503 + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_https_url_is_checked_before_sdk_allocates_connector(monkeypatch): + channel, _, _, factory = make_channel() + middleware = importlib.import_module( + "microsoft_agents.hosting.aiohttp.jwt_authorization_middleware", + ) + monkeypatch.setattr( + middleware, + "_authorize_request", + AsyncMock( + return_value=ClaimsIdentity( + {"aud": APP_ID, "iss": "https://api.botframework.com"}, + "JWT", + ) + ), + ) + try: + async with TestClient(TestServer(channel.create_app())) as client: + response = await client.post( + "/api/messages", + json=teams_event( + serviceUrl="http://smba.trafficmanager.net/amer/", + ), + ) + assert response.status == 400 + factory.create_connector_client.assert_not_called() + finally: + await channel.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "url", + [ + "http://smba.trafficmanager.net/", + "https://evil.example/", + "https://smba.trafficmanager.net.evil.example/", + "https://localhost/", + "https://user:password@smba.trafficmanager.net/", + ], +) +async def test_refuses_unsafe_service_urls(url): + channel, receipt, _, _ = make_channel() + source = normalize(teams_event(serviceUrl=url), app_id=APP_ID, tenant_id=TENANT_ID) + try: + with pytest.raises(ValueError, match="HTTPS Microsoft"): + await channel.reply(source, "answer") + assert not receipt.calls + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_on_turn_normalization_uses_sdk_aliases_and_missing_receipt_is_error(): + channel, _, _, _ = make_channel() + seen = [] + + async def receive(_, message): + seen.append(message) + + channel.bind(receive) + activity = Activity.model_validate(teams_event()) + await channel.on_turn(SimpleNamespace(activity=activity)) + await channel.close() + assert seen[0].text == "hello" + with pytest.raises(ValueError): + await channel.reply(replace(seen[0], metadata={}), "answer") diff --git a/tests/support/teams.py b/tests/support/teams.py new file mode 100644 index 0000000..7ef324f --- /dev/null +++ b/tests/support/teams.py @@ -0,0 +1,48 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from microsoft_agents.activity import ResourceResponse +from microsoft_agents.hosting.aiohttp import CloudAdapter +from microsoft_agents.hosting.core import OutboundHostValidator + +APP_ID = "11111111-1111-1111-1111-111111111111" +TENANT_ID = "22222222-2222-2222-2222-222222222222" + + +def teams_event(*, kind="personal", **changes): + event = { + "id": "incoming", + "type": "message", + "channelId": "msteams", + "text": "hello", + "serviceUrl": "https://smba.trafficmanager.net/amer/", + "from": {"id": "29:alice", "name": "Alice"}, + "recipient": {"id": f"28:{APP_ID}", "name": "Bot"}, + "conversation": {"id": "conversation", "conversationType": kind}, + "channelData": {"tenant": {"id": TENANT_ID}}, + } + event.update(changes) + return event + + +def teams_adapter(receipt): + async def send(*args): + return ResourceResponse.model_construct(id=receipt.get(*args)) + + connector = SimpleNamespace( + conversations=SimpleNamespace( + send_to_conversation=send, + reply_to_activity=send, + ), + close=AsyncMock(), + ) + token_client = SimpleNamespace(close=AsyncMock()) + factory = SimpleNamespace( + create_connector_client=AsyncMock(return_value=connector), + create_user_token_client=AsyncMock(return_value=token_client), + ) + adapter = CloudAdapter( + channel_service_client_factory=factory, + host_validator=OutboundHostValidator(enabled=True), + ) + return adapter, connector, factory From 9ee80c10d801bacc6ea01cb122041c2125092e17 Mon Sep 17 00:00:00 2001 From: moe-berri Date: Mon, 12 Oct 2026 14:50:29 -0400 Subject: [PATCH 4/6] feat: add iMessage through BlueBubbles Socket.IO and REST --- docs/imessage.md | 80 ++++++++ examples/imessage_echo.py | 26 +++ pyproject.toml | 2 + src/agentchat/channels/__init__.py | 5 +- src/agentchat/channels/imessage/__init__.py | 10 + src/agentchat/channels/imessage/adapter.py | 145 +++++++++++++++ src/agentchat/channels/imessage/events.py | 66 +++++++ src/agentchat/channels/imessage/rendering.py | 4 + src/agentchat/channels/imessage/transport.py | 89 +++++++++ tests/channels/imessage/test_imessage.py | 186 +++++++++++++++++++ tests/support/imessage.py | 51 +++++ 11 files changed, 662 insertions(+), 2 deletions(-) create mode 100644 docs/imessage.md create mode 100644 examples/imessage_echo.py create mode 100644 src/agentchat/channels/imessage/__init__.py create mode 100644 src/agentchat/channels/imessage/adapter.py create mode 100644 src/agentchat/channels/imessage/events.py create mode 100644 src/agentchat/channels/imessage/rendering.py create mode 100644 src/agentchat/channels/imessage/transport.py create mode 100644 tests/channels/imessage/test_imessage.py create mode 100644 tests/support/imessage.py diff --git a/docs/imessage.md b/docs/imessage.md new file mode 100644 index 0000000..58d431e --- /dev/null +++ b/docs/imessage.md @@ -0,0 +1,80 @@ +# iMessage through BlueBubbles + +Run AgentChat anywhere that can reach your BlueBubbles server. BlueBubbles itself +must run on a Mac signed into Messages. Replies are sent as that Mac's Apple +account, so use a dedicated account/Mac for an agent that should receive every +incoming conversation. A separate bot identity is not created. + +## Get the first reply + +1. Follow the [BlueBubbles server setup](https://docs.bluebubbles.app/server/installation-guides) + on the Mac. Grant its required Full Disk Access and Messages automation + permissions, and confirm the Mac can send and receive iMessages manually. + Set the BlueBubbles server password and enable its HTTP API/socket service. + This adapter's basic text replies use AppleScript and do not require the + BlueBubbles Private API or disabling System Integrity Protection. +2. Use the server's root URL, such as `http://127.0.0.1:1234` when AgentChat runs + on the same Mac. Use HTTPS for a remote server. A reverse proxy must forward + `/api/v1/` and `/socket.io/`, including WebSocket upgrades. No incoming public + webhook or Firebase configuration is needed by this adapter. +3. From this repository: + + ```bash + python -m pip install -e '.[imessage]' + export BLUEBUBBLES_URL='http://127.0.0.1:1234' + export BLUEBUBBLES_PASSWORD='your-server-password' + export IMESSAGE_INSTANCE_ID='agent-mac' + python examples/imessage_echo.py + ``` + + The stable instance ID keeps message/history identities unchanged if you move + the server URL. If omitted, it is derived from the URL. +4. From a different Apple account, send `hello` to the Mac's iMessage account. + Expect `hello` back. Sending from the same Apple account is ignored as an + outgoing/self message. Stop with Ctrl+C. + +For a test account that should only handle one conversation, restrict it before +starting: `export IMESSAGE_ALLOWED_CHATS='iMessage;-;+15551234567'`. Obtain the +actual chat GUID from BlueBubbles' chat API; do not infer it from a phone number. +The GUID above is an example. Comma-separate multiple GUIDs. Unset this variable +to receive all iMessage chats visible to the account. Programmatic +`allowed_chat_guids=[]` disables all conversations. + +## Routing and server behavior + +`IMessage(server_url=..., password=..., instance_id=...)` validates REST access, +then connects to authenticated Socket.IO `new-message` events. One chat GUID is +one conversation, including group chats; each message preserves its human sender. +Only iMessage chats are handled. SMS, outgoing messages, tapbacks/reactions, +service messages, and group-management events are ignored. + +Replies post to the same existing chat using `/api/v1/message/text`. The returned +message GUID is the receipt; `tempGuid` is only a correlation value, not an +idempotency guarantee. Native inline reply threads are treated as part of the +chat, and outgoing replies appear in its main timeline. There is no automatic +history backfill or replay of events missed while offline. Socket reconnection +recovers future traffic, not missed messages. + +Attachment-only messages retain file references, but downloads, uploads, rich +replies, and working indicators are not implemented. See [shared capabilities](providers.md). + +BlueBubbles authenticates using a password in request query parameters. The +adapter disables redirects, suppresses credential-bearing Socket.IO diagnostics, +and sanitizes REST errors. Use TLS remotely and exclude query strings from your +proxy access logs. Updating the server password requires updating the environment +and restarting the adapter. + +## Troubleshooting + +| Symptom | Check | +| --- | --- | +| REST HTTP 401 | BlueBubbles server password matches `BLUEBUBBLES_PASSWORD`. | +| REST succeeds but the socket disconnects | `/socket.io/` is forwarded, the password is current, and the server is running. | +| Incoming messages appear but sends fail | Messages is signed in; BlueBubbles can automate Messages and send through AppleScript. | +| No inbound event | Test from another Apple account; check the chat allowlist and that the Mac actually received the message. | +| Works only while the Mac is awake | Keep the BlueBubbles host running and reachable. | + +Local tests run real Socket.IO and REST round trips against a simulated server. +They do not exercise macOS Messages. The live echo above is the final check. + +Reference: [BlueBubbles REST API and events](https://docs.bluebubbles.app/server/developer-guides/rest-api-and-webhooks). diff --git a/examples/imessage_echo.py b/examples/imessage_echo.py new file mode 100644 index 0000000..9c1f878 --- /dev/null +++ b/examples/imessage_echo.py @@ -0,0 +1,26 @@ +"""Echo messages after completing docs/imessage.md; no model API key required.""" + +import asyncio +import logging + +from agentchat import AgentChat, MessageContext +from agentchat.channels.imessage import IMessage + + +async def main() -> None: + app = AgentChat(channels=[IMessage.from_env()]) + + @app.on_message + async def respond(context: MessageContext) -> str: + # Authorize the sender here before replacing this echo with an agent. + return context.message.text or "Received your attachment." + + try: + await app.run() + finally: + await app.close() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index ea66686..88837b3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,11 +23,13 @@ teams = [ "microsoft-agents-hosting-aiohttp>=1.8,<2", "microsoft-agents-authentication-msal>=1.8,<2", ] +imessage = ["python-socketio[asyncio-client]>=5.13,<6"] dev = [ "openai-agents>=0.4,<1", "discord.py>=2.6,<3", "microsoft-agents-hosting-aiohttp>=1.8,<2", "microsoft-agents-authentication-msal>=1.8,<2", + "python-socketio[asyncio-client]>=5.13,<6", "pytest>=7.3,<9", "pytest-asyncio>=0.21,<1", "ruff>=0.14,<1", diff --git a/src/agentchat/channels/__init__.py b/src/agentchat/channels/__init__.py index a0bf71d..03665d8 100644 --- a/src/agentchat/channels/__init__.py +++ b/src/agentchat/channels/__init__.py @@ -5,12 +5,13 @@ if TYPE_CHECKING: from agentchat.channels.discord import Discord + from agentchat.channels.imessage import IMessage from agentchat.channels.slack import Slack, SlackUser from agentchat.channels.teams import Teams -__all__ = ["Channel", "Discord", "Slack", "SlackUser", "Teams", "ThreadHistoryChannel"] +__all__ = ["Channel", "Discord", "IMessage", "Slack", "SlackUser", "Teams", "ThreadHistoryChannel"] -_PROVIDERS = {"Discord": "discord", "Teams": "teams"} +_PROVIDERS = {"Discord": "discord", "Teams": "teams", "IMessage": "imessage"} def __getattr__(name: str) -> object: diff --git a/src/agentchat/channels/imessage/__init__.py b/src/agentchat/channels/imessage/__init__.py new file mode 100644 index 0000000..f5aa302 --- /dev/null +++ b/src/agentchat/channels/imessage/__init__.py @@ -0,0 +1,10 @@ +try: + from agentchat.channels.imessage.adapter import IMessage +except ModuleNotFoundError as exc: + if exc.name == "socketio": + raise ImportError( + "Install iMessage support with: pip install 'agentchat[imessage]'" + ) from exc + raise + +__all__ = ["IMessage"] diff --git a/src/agentchat/channels/imessage/adapter.py b/src/agentchat/channels/imessage/adapter.py new file mode 100644 index 0000000..e9cb6ec --- /dev/null +++ b/src/agentchat/channels/imessage/adapter.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import asyncio +import hashlib +import logging +import os +from collections.abc import Collection +from types import MappingProxyType +from urllib.parse import quote + +import socketio + +from agentchat.channels._delivery import DeliveryTasks +from agentchat.channels.base import MessageReceiver +from agentchat.channels.imessage.events import normalize +from agentchat.channels.imessage.rendering import text_payload +from agentchat.channels.imessage.transport import BlueBubblesClient +from agentchat.models import Message, Sender + +logger = logging.getLogger(__name__) + + +class IMessage: + """iMessage through a BlueBubbles server, using Socket.IO and REST.""" + + name = "imessage" + + def __init__( + self, + *, + server_url: str, + password: str, + instance_id: str | None = None, + allowed_chat_guids: Collection[str] | None = None, + client: BlueBubblesClient | None = None, + socket_client: socketio.AsyncClient | None = None, + max_pending_events: int = 64, + shutdown_timeout: float = 20, + ) -> None: + if not password.strip(): + raise ValueError("BlueBubbles password is required") + self._client = client or BlueBubblesClient(server_url, password) + self._instance_id = ( + instance_id + or hashlib.sha256( + self._client.server_url.encode(), + ).hexdigest()[:24] + ) + self._allowed_chats = None if allowed_chat_guids is None else frozenset(allowed_chat_guids) + self._delivery = DeliveryTasks(max_pending_events, shutdown_timeout) + # Engine.IO can log full password-bearing URLs, even with logger=False. + socket_logger = logging.Logger(f"{__name__}.socket", level=logging.CRITICAL + 1) + socket_logger.propagate = False + self._socket = socket_client or socketio.AsyncClient( + reconnection=True, + logger=socket_logger, + engineio_logger=socket_logger, + handle_sigint=False, + ) + self._socket.on("new-message", self._on_message) + self._close_lock = asyncio.Lock() + self._closed = False + + @classmethod + def from_env(cls) -> IMessage: + url, password = os.getenv("BLUEBUBBLES_URL"), os.getenv("BLUEBUBBLES_PASSWORD") + if not url or not password: + raise RuntimeError("BLUEBUBBLES_URL and BLUEBUBBLES_PASSWORD are required") + chats = os.getenv("IMESSAGE_ALLOWED_CHATS") + return cls( + server_url=url, + password=password, + instance_id=os.getenv("IMESSAGE_INSTANCE_ID"), + allowed_chat_guids=[chat.strip() for chat in chats.split(",")] if chats else None, + ) + + def bind(self, receiver: MessageReceiver) -> None: + self._delivery.receiver = receiver + + async def run(self) -> None: + self._delivery.check_bound() + try: + await self._client.check_connection() + try: + await self._socket.connect(self._client.socket_url) + await self._socket.wait() + except Exception: + raise RuntimeError("BlueBubbles socket connection failed") from None + if not self._closed: + raise RuntimeError("BlueBubbles socket disconnected; check the server and password") + finally: + await self.close() + + async def close(self) -> None: + async with self._close_lock: + if self._closed: + return + self._closed = True + self._delivery.closed = True + try: + await self._socket.shutdown() + finally: + try: + await self._delivery.close() + finally: + await self._client.close() + + async def _on_message(self, event: object) -> None: + message = normalize(event, instance_id=self._instance_id) + if message is None or self._delivery.closed: + return + if ( + self._allowed_chats is not None + and message.metadata["chat_guid"] not in self._allowed_chats + ): + return + if not self._delivery.submit(self, message): + logger.warning("iMessage event dropped: max_pending_events reached") + + async def reply(self, source: Message, content: str) -> Message: + content = text_payload(content) + data = source.metadata + chat_guid = data.get("chat_guid") + if ( + source.channel != self.name + or data.get("instance_id") != self._instance_id + or not isinstance(chat_guid, str) + or not chat_guid.startswith("iMessage;") + ): + raise ValueError("Source does not belong to this iMessage adapter") + if self._allowed_chats is not None and chat_guid not in self._allowed_chats: + raise ValueError("iMessage conversation is not allowed") + self._delivery.check_reply() + receipt_id = await self._client.send_text(chat_guid, content) + if not isinstance(receipt_id, str) or not receipt_id: + raise RuntimeError("BlueBubbles did not return a message receipt") + return Message( + id=f"imessage:{quote(self._instance_id, safe='')}:{receipt_id}", + conversation_id=source.conversation_id, + channel=self.name, + sender=Sender(id=f"imessage:{self._instance_id}:me"), + text=content, + role="assistant", + metadata=MappingProxyType({**data, "message_guid": receipt_id}), + ) diff --git a/src/agentchat/channels/imessage/events.py b/src/agentchat/channels/imessage/events.py new file mode 100644 index 0000000..5305f80 --- /dev/null +++ b/src/agentchat/channels/imessage/events.py @@ -0,0 +1,66 @@ +"""BlueBubbles new-message events to SDK messages; no downloads or I/O.""" + +from __future__ import annotations + +from types import MappingProxyType +from urllib.parse import quote + +from agentchat.models import Attachment, Message, Sender + + +def normalize(event: object, *, instance_id: str) -> Message | None: + if not isinstance(event, dict) or event.get("isFromMe") is not False: + return None + if ( + event.get("isSystemMessage") + or event.get("isServiceMessage") + or event.get("isAutoReply") + or event.get("isSpam") + or event.get("associatedMessageType") not in (None, 0) + or event.get("itemType") not in (None, 0) + ): + return None + guid, handle, chats = event.get("guid"), event.get("handle"), event.get("chats") + if not isinstance(handle, dict) or not isinstance(chats, list) or len(chats) != 1: + return None + chat_guid = chats[0].get("guid") if isinstance(chats[0], dict) else None + address = handle.get("address") + if not all(isinstance(value, str) and value for value in (guid, chat_guid, address)): + return None + if not chat_guid.startswith("iMessage;"): + return None + text = event.get("text") or "" + if not isinstance(text, str): + return None + attachments = [] + items = event.get("attachments") + for item in items if isinstance(items, list) else []: + if isinstance(item, dict) and isinstance(item.get("guid"), str) and item["guid"]: + attachments.append( + Attachment( + id=item["guid"], + name=item.get("transferName"), + media_type=item.get("mimeType"), + size=item.get("totalBytes"), + ) + ) + text = text.replace("\ufffc", "").strip() + if not text and not attachments: + return None + scope = quote(instance_id, safe="") + return Message( + id=f"imessage:{scope}:{guid}", + conversation_id=f"imessage:{scope}:{quote(chat_guid, safe='')}", + channel="imessage", + sender=Sender(id=address), + text=text, + role="user", + attachments=tuple(attachments), + metadata=MappingProxyType( + { + "instance_id": instance_id, + "chat_guid": chat_guid, + "message_guid": guid, + } + ), + ) diff --git a/src/agentchat/channels/imessage/rendering.py b/src/agentchat/channels/imessage/rendering.py new file mode 100644 index 0000000..d37ae6d --- /dev/null +++ b/src/agentchat/channels/imessage/rendering.py @@ -0,0 +1,4 @@ +def text_payload(content: str) -> str: + if not isinstance(content, str) or not content.strip(): + raise ValueError("iMessage replies require nonempty text") + return content diff --git a/src/agentchat/channels/imessage/transport.py b/src/agentchat/channels/imessage/transport.py new file mode 100644 index 0000000..ead2f5b --- /dev/null +++ b/src/agentchat/channels/imessage/transport.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +from urllib.parse import quote +from uuid import uuid4 + +import aiohttp +from yarl import URL + + +class BlueBubblesClient: + """REST sends have no automatic retries; a timeout can mean an uncertain send.""" + + def __init__(self, server_url: str, password: str) -> None: + url = URL(server_url) + if ( + url.scheme not in ("http", "https") + or not url.host + or url.user + or url.password + or url.query_string + or url.fragment + or url.path not in ("", "/") + ): + raise ValueError("BlueBubbles requires a root server URL without credentials or query") + if url.scheme == "http" and url.host not in ("localhost", "127.0.0.1", "::1"): + raise ValueError("Use HTTPS for a remote BlueBubbles server") + self.server_url = str(url).rstrip("/") + self._password = password + self._session: aiohttp.ClientSession | None = None + self._closed = False + + @property + def socket_url(self) -> str: + # BlueBubbles applies decodeURI after Socket.IO parses the query string. + # Escape percent itself for that second decode, but retain reserved characters. + password = self._password.replace("%", "%25") + return f"{self.server_url}?password={quote(password, safe='')}" + + async def request(self, method: str, path: str, **kwargs: object) -> dict: + if self._closed: + raise RuntimeError("BlueBubbles client is closed") + if self._session is None: + self._session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) + try: + async with self._session.request( + method, + f"{self.server_url}{path}", + params={"password": self._password}, + allow_redirects=False, + **kwargs, + ) as response: + if not 200 <= response.status < 300: + raise RuntimeError(f"BlueBubbles request failed (HTTP {response.status})") + result = await response.json() + if not isinstance(result, dict) or result.get("status") != 200: + raise RuntimeError("BlueBubbles returned an invalid response") + return result + except (aiohttp.ClientError, TimeoutError, ValueError): + # aiohttp exception strings may include the password-bearing request URL. + raise RuntimeError("BlueBubbles request failed; delivery may be uncertain") from None + + async def check_connection(self) -> None: + await self.request("GET", "/api/v1/ping") + + async def send_text(self, chat_guid: str, content: str) -> str: + result = await self.request( + "POST", + "/api/v1/message/text", + json={ + "chatGuid": chat_guid, + "message": content, + "method": "apple-script", + "tempGuid": str(uuid4()), + }, + ) + data = result.get("data") + if ( + not isinstance(data, dict) + or not isinstance(data.get("guid"), str) + or not data["guid"] + or data.get("error") not in (None, 0) + ): + raise RuntimeError("BlueBubbles did not return a successful message receipt") + return data["guid"] + + async def close(self) -> None: + self._closed = True + if self._session: + await self._session.close() diff --git a/tests/channels/imessage/test_imessage.py b/tests/channels/imessage/test_imessage.py new file mode 100644 index 0000000..57f826b --- /dev/null +++ b/tests/channels/imessage/test_imessage.py @@ -0,0 +1,186 @@ +import asyncio +from urllib.parse import parse_qs + +import pytest +import socketio +from aiohttp import web +from aiohttp.test_utils import TestServer + +from agentchat import AgentChat +from agentchat.channels.imessage import IMessage +from agentchat.channels.imessage.events import normalize +from agentchat.channels.imessage.transport import BlueBubblesClient +from tests.support.imessage import CHAT_GUID, BubblesClient, SocketClient, imessage_event +from tests.support.receipts import Receipt + + +@pytest.mark.parametrize( + "changes", + [ + {"isFromMe": True}, + {"isFromMe": None}, + {"associatedMessageType": 2000}, + {"isSystemMessage": True}, + {"itemType": 1}, + {"handle": None}, + {"chats": []}, + {"chats": [{"guid": "SMS;-;+15551234567"}]}, + {"text": ""}, + {"chats": [None]}, + {"guid": ""}, + {"attachments": 7, "text": ""}, + ], +) +def test_filters_echoes_reactions_system_sms_and_malformed_events(changes): + assert normalize(imessage_event(**changes), instance_id="mac") is None + + +def test_attachment_only_and_group_sender_identity(): + event = imessage_event( + text="\ufffc", + chats=[{"guid": "iMessage;+;group"}], + attachments=[ + { + "guid": "file", + "transferName": "photo.jpg", + "mimeType": "image/jpeg", + "totalBytes": 50, + } + ], + ) + message = normalize(event, instance_id="mac") + assert message.sender.id == "+15551234567" + assert message.text == "" + assert message.attachments[0].size == 50 + assert message.conversation_id != normalize(imessage_event(), instance_id="mac").conversation_id + assert message.id != normalize(event, instance_id="other-mac").id + + +@pytest.mark.asyncio +async def test_allowlist_applies_to_receiving_and_sending(): + receipt = Receipt() + channel = IMessage( + server_url="http://localhost:1234", + password="fake", + instance_id="mac", + client=BubblesClient(receipt), + socket_client=SocketClient(), + allowed_chat_guids=[], + ) + seen = [] + + async def receive(_, message): + seen.append(message) + + channel.bind(receive) + await channel._on_message(imessage_event()) + with pytest.raises(ValueError, match="not allowed"): + await channel.reply(normalize(imessage_event(), instance_id="mac"), "answer") + await channel.close() + assert not seen and not receipt.calls + + +@pytest.mark.asyncio +async def test_real_socketio_and_rest_round_trip(): + server_socket = socketio.AsyncServer(async_mode="aiohttp") + server_app = web.Application() + server_socket.attach(server_app) + sends = [] + password = "local-test &secret" + connected, sent = asyncio.Event(), asyncio.Event() + + @server_socket.event + async def connect(sid, environ, auth): + assert parse_qs(environ["QUERY_STRING"])["password"] == [password] + connected.set() + + async def ping(request): + assert request.query["password"] == password + return web.json_response({"status": 200, "data": "pong"}) + + async def send(request): + assert request.query["password"] == password + sends.append(await request.json()) + sent.set() + return web.json_response({"status": 200, "data": {"guid": "sent-guid", "error": 0}}) + + server_app.router.add_get("/api/v1/ping", ping) + server_app.router.add_post("/api/v1/message/text", send) + async with TestServer(server_app) as server: + channel = IMessage( + server_url=str(server.make_url("/")), password=password, instance_id="mac" + ) + app = AgentChat(channels=[channel]) + + @app.on_message + async def respond(context): + return "echo: " + context.message.text + + running = asyncio.create_task(app.run()) + try: + await asyncio.wait_for(connected.wait(), timeout=3) + await server_socket.emit("new-message", imessage_event()) + await asyncio.wait_for(sent.wait(), timeout=3) + await channel.close() + await asyncio.wait_for(running, timeout=3) + assert sends[0]["message"] == "echo: hello" + assert sends[0]["chatGuid"] == CHAT_GUID + assert sends[0]["method"] == "apple-script" + assert sends[0]["tempGuid"] != "sent-guid" + source = normalize(imessage_event(), instance_id="mac") + history = await app.state.history(source.conversation_id) + assert len(history) == 2 + assert history[-1].id == "imessage:mac:sent-guid" + assert channel._client._session.closed + assert not channel._socket.connected + finally: + await app.close() + running.cancel() + await asyncio.gather(running, return_exceptions=True) + await server_socket.shutdown() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "status,data", + [ + (401, {"message": "contains-secret"}), + (302, {}), + (200, {"status": 200, "data": {"tempGuid": "not-a-receipt"}}), + (200, {"status": 200, "data": {"guid": "sent", "error": 4}}), + ], +) +async def test_http_errors_redirects_and_unconfirmed_sends(status, data): + count = 0 + + async def send(request): + nonlocal count + count += 1 + return web.json_response(data, status=status, headers={"Location": "/never-follow"}) + + server_app = web.Application() + server_app.router.add_post("/api/v1/message/text", send) + async with TestServer(server_app) as server: + client = BlueBubblesClient(str(server.make_url("/")), "contains-secret") + try: + with pytest.raises(RuntimeError) as error: + await client.send_text(CHAT_GUID, "hello") + assert "contains-secret" not in str(error.value) + assert count == 1 + finally: + await client.close() + + +@pytest.mark.parametrize( + "url", + [ + "http://example.com", + "https://user:password@example.com", + "https://example.com?password=x", + "https://example.com/path", + "ftp://example.com", + ], +) +def test_url_validation(url): + with pytest.raises(ValueError): + BlueBubblesClient(url, "secret") diff --git a/tests/support/imessage.py b/tests/support/imessage.py new file mode 100644 index 0000000..af0679f --- /dev/null +++ b/tests/support/imessage.py @@ -0,0 +1,51 @@ +import asyncio + +CHAT_GUID = "iMessage;-;+15551234567" + + +def imessage_event(**changes): + return { + "guid": "incoming", + "text": "hello", + "isFromMe": False, + "itemType": 0, + "associatedMessageType": 0, + "handle": {"address": "+15551234567"}, + "chats": [{"guid": CHAT_GUID}], + **changes, + } + + +class BubblesClient: + server_url = "http://localhost:1234" + socket_url = "http://localhost:1234?password=fake" + + def __init__(self, receipt): + self.receipt = receipt + self.closed = False + + async def check_connection(self): + pass + + async def send_text(self, *args): + return self.receipt.get(*args) + + async def close(self): + self.closed = True + + +class SocketClient: + def __init__(self): + self.stopped, self.started = asyncio.Event(), asyncio.Event() + + def on(self, event, callback): + self.callback = callback + + async def connect(self, url): + self.started.set() + + async def wait(self): + await self.stopped.wait() + + async def shutdown(self): + self.stopped.set() From 34a5d05edd393957139eff5ff5d31e3e2e654a9a Mon Sep 17 00:00:00 2001 From: moe-berri Date: Mon, 12 Oct 2026 14:50:56 -0400 Subject: [PATCH 5/6] feat: add signed WhatsApp Cloud API webhooks and text replies --- docs/whatsapp.md | 90 +++++++++ examples/whatsapp_echo.py | 26 +++ pyproject.toml | 1 + src/agentchat/channels/__init__.py | 23 ++- src/agentchat/channels/whatsapp/__init__.py | 3 + src/agentchat/channels/whatsapp/adapter.py | 196 ++++++++++++++++++ src/agentchat/channels/whatsapp/events.py | 91 +++++++++ src/agentchat/channels/whatsapp/rendering.py | 13 ++ tests/channels/whatsapp/test_whatsapp.py | 202 +++++++++++++++++++ tests/support/whatsapp.py | 51 +++++ 10 files changed, 692 insertions(+), 4 deletions(-) create mode 100644 docs/whatsapp.md create mode 100644 examples/whatsapp_echo.py create mode 100644 src/agentchat/channels/whatsapp/__init__.py create mode 100644 src/agentchat/channels/whatsapp/adapter.py create mode 100644 src/agentchat/channels/whatsapp/events.py create mode 100644 src/agentchat/channels/whatsapp/rendering.py create mode 100644 tests/channels/whatsapp/test_whatsapp.py create mode 100644 tests/support/whatsapp.py diff --git a/docs/whatsapp.md b/docs/whatsapp.md new file mode 100644 index 0000000..06792e7 --- /dev/null +++ b/docs/whatsapp.md @@ -0,0 +1,90 @@ +# WhatsApp Business Cloud API + +Connect an official WhatsApp Business phone number through Meta's Cloud API. +This adapter does not automate a personal WhatsApp account or require WhatsApp +Web/QR pairing. For an initial test, use the test phone number supplied by Meta. + +## Get the first reply + +1. In [Meta for Developers](https://developers.facebook.com/apps/), create/configure + an app with the WhatsApp product/use case and connect a business portfolio. + Open **WhatsApp → API Setup**, select the test number, and add your own WhatsApp + account as a verified test recipient. Record the **Phone number ID** and the + temporary access token. The ID is not the visible phone number or business ID. +2. Under the app's **Settings → Basic**, obtain the **App secret**. Choose a separate + random webhook verification token, for example with + `python -c 'import secrets; print(secrets.token_urlsafe(32))'`. +3. From this repository, install and start the server: + + ```bash + python -m pip install -e '.[whatsapp]' + export WHATSAPP_PHONE_NUMBER_ID='numeric-phone-number-id' + export WHATSAPP_ACCESS_TOKEN='your-access-token' + export WHATSAPP_APP_SECRET='your-meta-app-secret' + export WHATSAPP_VERIFY_TOKEN='your-chosen-verification-token' + python examples/whatsapp_echo.py + ``` + + It listens on `127.0.0.1:3979`. Forward a public HTTPS tunnel/proxy to this port. + Containers can set `WHATSAPP_HOST=0.0.0.0`; `WHATSAPP_PORT` changes the port. +4. In WhatsApp's webhook configuration, enter + `https://YOUR_HOST/webhooks/whatsapp` as the callback URL and the same + verification token. Verify/save, then subscribe to the **messages** field + for the connected WhatsApp Business Account. Keep the app subscribed to that + account when moving from a test number to a production number. +5. From the verified test recipient, send `hello` to the test business number. + Expect `hello` back. Sending a user message opens the customer-service window + needed for a free-form text reply. Stop with Ctrl+C. + +## Routing and limits + +Each `(phone_number_id, sender WhatsApp ID)` pair is a conversation. Signed webhook +batches can contain multiple inputs; sender profile names are matched by WhatsApp +ID. Events for other phone-number IDs, status receipts, reactions, system events, +and unsupported message types do not invoke the agent. Text, button/list reply +titles, and image/audio/video/document/sticker references are normalized. + +`WhatsApp(phone_number_id=..., access_token=..., app_secret=..., verify_token=...)` +checks the raw-body `X-Hub-Signature-256` HMAC before parsing messages. The webhook +verification token is for the GET challenge; the app secret authenticates POSTs. +The reverse proxy must preserve the request body and signature header exactly. + +Replies quote the original message through `context.message_id`, disable link +previews, and require nonempty text of at most 4,096 characters. They use Graph +API `v23.0` by default; set `WHATSAPP_API_VERSION` to another supported version +when upgrading. HTTP errors, rate limits, and timeouts propagate without retries. +A returned `wamid` confirms API acceptance, not handset delivery/read status. + +Free-form replies are subject to Meta's customer-service window, generally 24 +hours from the user's last message. Template messages, proactive campaigns, +group conversations, media downloads/uploads, and delivery-status callbacks are +outside this adapter's current interface. Do not use a stored source message to +assume the text-reply window remains open. + +For an existing aiohttp server, host `whatsapp.create_app()` yourself and bind the +receiver before serving requests. This exposes the same verification and signed +webhook handlers. Call `await whatsapp.close()` at shutdown; do not also call +`whatsapp.run()`. See [shared usage and delivery semantics](providers.md). + +## Production setup and troubleshooting + +Replace Meta's temporary token with a system-user token that has access to the +registered business phone number and `whatsapp_business_messaging` permission. +Business/account setup and subscription operations can also require +`whatsapp_business_management`. Complete Meta's required business/number +registration and app access steps before serving real users. Store tokens as +secrets, monitor expiry/revocation, and restart after rotation. + +| Symptom | Check | +| --- | --- | +| Webhook verification fails | HTTPS URL/path, server is running, and matching `WHATSAPP_VERIFY_TOKEN`. | +| POST returns 401 | Correct Meta app secret and an unmodified raw body/signature header. | +| Webhook verifies but no messages arrive | Subscribe to `messages` on the correct business account and check the phone-number ID. | +| Send returns HTTP 400/401/403 | Token expiry/permissions, phone-number ID, test recipient verification, and an open customer-service window. | +| HTTP 503 from this receiver | Its pending-work limit is full; see the shared delivery guide. | + +Local tests exercise signed webhooks and actual HTTP sends to a simulated Graph +endpoint. They do not establish that your Meta account is provisioned correctly. +Verify the live echo before connecting an agent. + +Reference: [WhatsApp Cloud API documentation](https://developers.facebook.com/docs/whatsapp/cloud-api/). diff --git a/examples/whatsapp_echo.py b/examples/whatsapp_echo.py new file mode 100644 index 0000000..8879b8b --- /dev/null +++ b/examples/whatsapp_echo.py @@ -0,0 +1,26 @@ +"""Echo messages after completing docs/whatsapp.md; no model API key required.""" + +import asyncio +import logging + +from agentchat import AgentChat, MessageContext +from agentchat.channels.whatsapp import WhatsApp + + +async def main() -> None: + app = AgentChat(channels=[WhatsApp.from_env()]) + + @app.on_message + async def respond(context: MessageContext) -> str: + # Authorize the sender here before replacing this echo with an agent. + return context.message.text or "Received your attachment." + + try: + await app.run() + finally: + await app.close() + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 88837b3..385d9a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,7 @@ teams = [ "microsoft-agents-authentication-msal>=1.8,<2", ] imessage = ["python-socketio[asyncio-client]>=5.13,<6"] +whatsapp = [] # Uses the existing aiohttp dependency. dev = [ "openai-agents>=0.4,<1", "discord.py>=2.6,<3", diff --git a/src/agentchat/channels/__init__.py b/src/agentchat/channels/__init__.py index 03665d8..3f47735 100644 --- a/src/agentchat/channels/__init__.py +++ b/src/agentchat/channels/__init__.py @@ -8,10 +8,25 @@ from agentchat.channels.imessage import IMessage from agentchat.channels.slack import Slack, SlackUser from agentchat.channels.teams import Teams - -__all__ = ["Channel", "Discord", "IMessage", "Slack", "SlackUser", "Teams", "ThreadHistoryChannel"] - -_PROVIDERS = {"Discord": "discord", "Teams": "teams", "IMessage": "imessage"} + from agentchat.channels.whatsapp import WhatsApp + +__all__ = [ + "Channel", + "Discord", + "IMessage", + "Slack", + "SlackUser", + "Teams", + "ThreadHistoryChannel", + "WhatsApp", +] + +_PROVIDERS = { + "Discord": "discord", + "Teams": "teams", + "IMessage": "imessage", + "WhatsApp": "whatsapp", +} def __getattr__(name: str) -> object: diff --git a/src/agentchat/channels/whatsapp/__init__.py b/src/agentchat/channels/whatsapp/__init__.py new file mode 100644 index 0000000..057dfde --- /dev/null +++ b/src/agentchat/channels/whatsapp/__init__.py @@ -0,0 +1,3 @@ +from agentchat.channels.whatsapp.adapter import WhatsApp + +__all__ = ["WhatsApp"] diff --git a/src/agentchat/channels/whatsapp/adapter.py b/src/agentchat/channels/whatsapp/adapter.py new file mode 100644 index 0000000..925d932 --- /dev/null +++ b/src/agentchat/channels/whatsapp/adapter.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +import asyncio +import hashlib +import hmac +import json +import os +import re +from types import MappingProxyType + +import aiohttp +from aiohttp import web + +from agentchat.channels._delivery import DeliveryTasks +from agentchat.channels.base import MessageReceiver +from agentchat.channels.whatsapp.events import normalize +from agentchat.channels.whatsapp.rendering import text_payload +from agentchat.models import Message, Sender + +GRAPH_ROOT = "https://graph.facebook.com" + + +class WhatsApp: + """WhatsApp Business Cloud API: signed webhooks and text replies.""" + + name = "whatsapp" + + def __init__( + self, + *, + phone_number_id: str, + access_token: str, + app_secret: str, + verify_token: str, + api_version: str = "v23.0", + host: str = "127.0.0.1", + port: int = 3979, + max_pending_events: int = 64, + shutdown_timeout: float = 20, + ) -> None: + if not all( + value.strip() for value in (phone_number_id, access_token, app_secret, verify_token) + ): + raise ValueError( + "WhatsApp phone_number_id, access_token, app_secret, verify_token required" + ) + if not re.fullmatch(r"[0-9]+", phone_number_id): + raise ValueError("WhatsApp phone_number_id must be the numeric API ID") + if not re.fullmatch(r"v[1-9][0-9]*\.[0-9]+", api_version): + raise ValueError("WhatsApp api_version must have the form v23.0") + self._phone_number_id, self._access_token = phone_number_id, access_token + self._app_secret, self._verify_token = app_secret, verify_token + self._api_version, self._host, self._port = api_version, host, port + self._delivery = DeliveryTasks(max_pending_events, shutdown_timeout) + self._session: aiohttp.ClientSession | None = None + self._runner: web.AppRunner | None = None + self._stopped = asyncio.Event() + self._close_lock = asyncio.Lock() + self._closed = False + + @classmethod + def from_env(cls) -> WhatsApp: + keys = ( + "WHATSAPP_PHONE_NUMBER_ID", + "WHATSAPP_ACCESS_TOKEN", + "WHATSAPP_APP_SECRET", + "WHATSAPP_VERIFY_TOKEN", + ) + values = [os.getenv(key) for key in keys] + if not all(values): + raise RuntimeError(f"{', '.join(keys)} are required") + return cls( + phone_number_id=values[0], + access_token=values[1], + app_secret=values[2], + verify_token=values[3], + api_version=os.getenv("WHATSAPP_API_VERSION", "v23.0"), + host=os.getenv("WHATSAPP_HOST", "127.0.0.1"), + port=int(os.getenv("WHATSAPP_PORT", "3979")), + ) + + def bind(self, receiver: MessageReceiver) -> None: + self._delivery.receiver = receiver + + def create_app(self) -> web.Application: + app = web.Application() + app.router.add_get("/webhooks/whatsapp", self._verify) + app.router.add_post("/webhooks/whatsapp", self._receive) + return app + + async def run(self) -> None: + self._delivery.check_bound() + self._runner = web.AppRunner(self.create_app(), access_log=None) + try: + await self._runner.setup() + await web.TCPSite(self._runner, self._host, self._port).start() + await self._stopped.wait() + finally: + await self.close() + + async def close(self) -> None: + async with self._close_lock: + if self._closed: + return + self._closed = True + self._delivery.closed = True + self._stopped.set() + try: + if self._runner: + await self._runner.cleanup() + finally: + try: + await self._delivery.close() + finally: + if self._session: + await self._session.close() + + async def _verify(self, request: web.Request) -> web.Response: + query = request.query + if ( + query.get("hub.mode") != "subscribe" + or not query.get("hub.challenge") + or not hmac.compare_digest( + query.get("hub.verify_token", "").encode(), + self._verify_token.encode(), + ) + ): + raise web.HTTPForbidden(text="Webhook verification failed") + return web.Response(text=query["hub.challenge"]) + + async def _receive(self, request: web.Request) -> web.Response: + raw = await request.read() + signature = "sha256=" + hmac.new(self._app_secret.encode(), raw, hashlib.sha256).hexdigest() + supplied = request.headers.get("X-Hub-Signature-256", "") + if not hmac.compare_digest(supplied.encode(), signature.encode()): + raise web.HTTPUnauthorized(text="Invalid webhook signature") + try: + payload = json.loads(raw) + except (ValueError, UnicodeError): + raise web.HTTPBadRequest(text="Invalid webhook JSON") from None + messages = normalize(payload, phone_number_id=self._phone_number_id) + # Admit a batch atomically so a 503 never follows a partially accepted batch. + if ( + self._delivery.closed + or len(self._delivery.tasks) + len(messages) > self._delivery.limit + ): + raise web.HTTPServiceUnavailable(text="WhatsApp receiver is busy") + for message in messages: + self._delivery.submit(self, message) + return web.Response(text="OK") + + async def reply(self, source: Message, content: str) -> Message: + data = source.metadata + recipient, message_id = data.get("wa_id"), data.get("message_id") + if ( + source.channel != self.name + or data.get("phone_number_id") != self._phone_number_id + or not isinstance(recipient, str) + or not re.fullmatch(r"[0-9]+", recipient) + or not isinstance(message_id, str) + or not message_id + ): + raise ValueError("Source does not belong to this WhatsApp adapter") + payload = text_payload(content, recipient=recipient, message_id=message_id) + self._delivery.check_reply() + if self._session is None: + self._session = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) + try: + async with self._session.post( + f"{GRAPH_ROOT}/{self._api_version}/{self._phone_number_id}/messages", + headers={"Authorization": f"Bearer {self._access_token}"}, + json=payload, + allow_redirects=False, + ) as response: + if not 200 <= response.status < 300: + raise RuntimeError(f"WhatsApp send failed (HTTP {response.status})") + result = await response.json() + except (aiohttp.ClientError, TimeoutError, ValueError): + raise RuntimeError("WhatsApp request failed; delivery may be uncertain") from None + receipts = result.get("messages") if isinstance(result, dict) else None + receipt_id = ( + receipts[0].get("id") + if (isinstance(receipts, list) and len(receipts) == 1 and isinstance(receipts[0], dict)) + else None + ) + if not isinstance(receipt_id, str) or not receipt_id: + raise RuntimeError("WhatsApp did not return a message receipt") + return Message( + id=f"whatsapp:{self._phone_number_id}:{receipt_id}", + conversation_id=source.conversation_id, + channel=self.name, + sender=Sender(id=self._phone_number_id), + text=content, + role="assistant", + metadata=MappingProxyType({**data, "message_id": receipt_id}), + ) diff --git a/src/agentchat/channels/whatsapp/events.py b/src/agentchat/channels/whatsapp/events.py new file mode 100644 index 0000000..9ca4085 --- /dev/null +++ b/src/agentchat/channels/whatsapp/events.py @@ -0,0 +1,91 @@ +"""Normalize signed WhatsApp Cloud API webhook batches.""" + +from __future__ import annotations + +from types import MappingProxyType + +from agentchat.models import Attachment, Message, Sender + + +def _mapping(value: object) -> dict: + return value if isinstance(value, dict) else {} + + +def _items(value: object) -> list: + return value if isinstance(value, list) else [] + + +def normalize(payload: object, *, phone_number_id: str) -> tuple[Message, ...]: + payload = _mapping(payload) + if payload.get("object") != "whatsapp_business_account": + return () + messages = [] + for entry in _items(payload.get("entry")): + for change in _items(_mapping(entry).get("changes")): + change = _mapping(change) + value = _mapping(change.get("value")) + if ( + change.get("field") != "messages" + or value.get("messaging_product") != "whatsapp" + or _mapping(value.get("metadata")).get("phone_number_id") != phone_number_id + ): + continue + names = { + item.get("wa_id"): _mapping(item.get("profile")).get("name") + for item in _items(value.get("contacts")) + if isinstance(item, dict) and isinstance(item.get("wa_id"), str) + } + for raw in _items(value.get("messages")): + event = _mapping(raw) + sender, message_id = event.get("from"), event.get("id") + if ( + not isinstance(sender, str) + or not sender.isascii() + or not sender.isdigit() + or not isinstance(message_id, str) + or not message_id + ): + continue + kind = event.get("type") + text, attachments = "", () + if kind == "text": + text = _mapping(event.get("text")).get("body", "") + elif kind in ("image", "audio", "video", "document", "sticker"): + media = _mapping(event.get(kind)) + if isinstance(media.get("id"), str) and media["id"]: + attachments = ( + Attachment( + id=media["id"], + name=media.get("filename"), + media_type=media.get("mime_type"), + ), + ) + text = media.get("caption", "") + elif kind == "button": + text = _mapping(event.get("button")).get("text", "") + elif kind == "interactive": + interactive = _mapping(event.get("interactive")) + reply = interactive.get("button_reply") or interactive.get("list_reply") + text = _mapping(reply).get("title", "") + if not isinstance(text, str) or (not text.strip() and not attachments): + continue + messages.append( + Message( + id=f"whatsapp:{phone_number_id}:{message_id}", + conversation_id=f"whatsapp:{phone_number_id}:{sender}", + channel="whatsapp", + sender=Sender(id=sender, display_name=names.get(sender)), + text=text, + role="user", + attachments=attachments, + metadata=MappingProxyType( + { + "phone_number_id": phone_number_id, + "message_id": message_id, + "wa_id": sender, + "message_type": kind, + } + ), + ) + ) + return tuple(messages) diff --git a/src/agentchat/channels/whatsapp/rendering.py b/src/agentchat/channels/whatsapp/rendering.py new file mode 100644 index 0000000..c4eb7c3 --- /dev/null +++ b/src/agentchat/channels/whatsapp/rendering.py @@ -0,0 +1,13 @@ +def text_payload(content: str, *, recipient: str, message_id: str) -> dict: + if not isinstance(content, str) or not content.strip(): + raise ValueError("WhatsApp replies require nonempty text") + if len(content) > 4096: + raise ValueError("WhatsApp replies must fit 4,096 characters; send separate replies") + return { + "messaging_product": "whatsapp", + "recipient_type": "individual", + "to": recipient, + "type": "text", + "text": {"body": content, "preview_url": False}, + "context": {"message_id": message_id}, + } diff --git a/tests/channels/whatsapp/test_whatsapp.py b/tests/channels/whatsapp/test_whatsapp.py new file mode 100644 index 0000000..26d2ce2 --- /dev/null +++ b/tests/channels/whatsapp/test_whatsapp.py @@ -0,0 +1,202 @@ +import asyncio +import hashlib +import hmac +import json + +import pytest +from aiohttp import web +from aiohttp.test_utils import TestClient, TestServer + +from agentchat import AgentChat +from agentchat.channels.whatsapp import WhatsApp +from agentchat.channels.whatsapp.events import normalize +from agentchat.channels.whatsapp.rendering import text_payload +from tests.support.receipts import Receipt +from tests.support.whatsapp import WhatsAppSession, whatsapp_event + + +def make_channel(**options): + return WhatsApp( + phone_number_id="123", + access_token="token", + app_secret="secret", + verify_token="verify", + **options, + ) + + +def signed(payload): + raw = json.dumps(payload).encode() + signature = hmac.new(b"secret", raw, hashlib.sha256).hexdigest() + return {"data": raw, "headers": {"X-Hub-Signature-256": f"sha256={signature}"}} + + +@pytest.mark.asyncio +async def test_verification_and_rejects_unsigned_or_tampered_webhooks(): + channel = make_channel() + try: + async with TestClient(TestServer(channel.create_app())) as client: + response = await client.get( + "/webhooks/whatsapp", + params={ + "hub.mode": "subscribe", + "hub.verify_token": "verify", + "hub.challenge": "12345", + }, + ) + assert response.status == 200 and await response.text() == "12345" + response = await client.get( + "/webhooks/whatsapp", + params={ + "hub.mode": "subscribe", + "hub.verify_token": "wrong", + "hub.challenge": "12345", + }, + ) + assert response.status == 403 + response = await client.post("/webhooks/whatsapp", json=whatsapp_event()) + assert response.status == 401 + payload = signed(whatsapp_event()) + payload["data"] += b" " + response = await client.post("/webhooks/whatsapp", **payload) + assert response.status == 401 + finally: + await channel.close() + + +@pytest.mark.asyncio +async def test_real_webhook_to_graph_round_trip_and_duplicate_claim(monkeypatch): + sends = [] + sent = asyncio.Event() + + async def send(request): + assert request.headers["Authorization"] == "Bearer token" + sends.append(await request.json()) + sent.set() + return web.json_response( + {"messaging_product": "whatsapp", "messages": [{"id": "wamid.sent"}]} + ) + + graph = web.Application() + graph.router.add_post("/v23.0/123/messages", send) + async with TestServer(graph) as graph_server: + monkeypatch.setattr( + "agentchat.channels.whatsapp.adapter.GRAPH_ROOT", + str( + graph_server.make_url("/"), + ).rstrip("/"), + ) + channel = make_channel() + app = AgentChat(channels=[channel]) + + @app.on_message + async def respond(context): + return "echo: " + context.message.text + + try: + async with TestClient(TestServer(channel.create_app())) as client: + response = await client.post("/webhooks/whatsapp", **signed(whatsapp_event())) + assert response.status == 200 + await asyncio.wait_for(sent.wait(), timeout=2) + response = await client.post("/webhooks/whatsapp", **signed(whatsapp_event())) + assert response.status == 200 + await app.close() + assert len(sends) == 1 + assert sends[0]["context"] == {"message_id": "wamid.incoming"} + assert sends[0]["to"] == "15551234567" + assert sends[0]["text"] == {"body": "echo: hello", "preview_url": False} + history = await app.state.history("whatsapp:123:15551234567") + assert history[-1].id == "whatsapp:123:wamid.sent" + finally: + await app.close() + + +@pytest.mark.asyncio +async def test_batch_overload_never_partially_accepts(): + channel = make_channel(max_pending_events=1) + seen = [] + + async def receive(*args): + seen.append(args) + + channel.bind(receive) + messages = [ + {"id": f"wamid.{i}", "from": "15551234567", "type": "text", "text": {"body": "hello"}} + for i in range(2) + ] + try: + async with TestClient(TestServer(channel.create_app())) as client: + response = await client.post( + "/webhooks/whatsapp", **signed(whatsapp_event(messages=messages)) + ) + assert response.status == 503 + assert not seen and not channel._delivery.tasks + finally: + await channel.close() + + +def test_batches_media_button_inputs_and_account_isolation(): + messages = [ + { + "id": "image", + "from": "15551234567", + "type": "image", + "image": {"id": "media-id", "mime_type": "image/png"}, + }, + { + "id": "button", + "from": "15557654321", + "type": "interactive", + "interactive": {"button_reply": {"id": "yes", "title": "Yes"}}, + }, + {"id": "reaction", "from": "15551234567", "type": "reaction"}, + ] + output = normalize(whatsapp_event(messages=messages), phone_number_id="123") + assert len(output) == 2 + assert output[0].attachments[0].id == "media-id" + assert output[1].text == "Yes" + assert output[0].conversation_id != output[1].conversation_id + assert output[0].sender.display_name == "Alice" + assert output[1].sender.display_name is None + assert normalize(whatsapp_event(), phone_number_id="other") == () + + +@pytest.mark.parametrize( + "payload", + [ + None, + [], + {"object": "wrong"}, + { + "object": "whatsapp_business_account", + "entry": [{"changes": [None]}], + }, + ], +) +def test_malformed_and_status_only_events_are_ignored(payload): + assert normalize(payload, phone_number_id="123") == () + + +def test_whatsapp_text_limit(): + assert text_payload("a" * 4096, recipient="123", message_id="m")["type"] == "text" + with pytest.raises(ValueError, match="4,096"): + text_payload("a" * 4097, recipient="123", message_id="m") + + +@pytest.mark.asyncio +async def test_shutdown_drains_handler_before_first_http_session_is_created(monkeypatch): + receipt = Receipt() + session = WhatsAppSession(receipt) + monkeypatch.setattr( + "agentchat.channels.whatsapp.adapter.aiohttp.ClientSession", lambda **kw: session + ) + channel = make_channel() + + async def receive(channel, source): + await channel.reply(source, "final reply") + + channel.bind(receive) + channel._delivery.submit(channel, normalize(whatsapp_event(), phone_number_id="123")[0]) + await channel.close() + assert len(receipt.calls) == 1 + assert session.closed diff --git a/tests/support/whatsapp.py b/tests/support/whatsapp.py new file mode 100644 index 0000000..767da3d --- /dev/null +++ b/tests/support/whatsapp.py @@ -0,0 +1,51 @@ +def whatsapp_event(*, messages=None, phone_number_id="123"): + return { + "object": "whatsapp_business_account", + "entry": [ + { + "id": "business", + "changes": [ + { + "field": "messages", + "value": { + "messaging_product": "whatsapp", + "metadata": {"phone_number_id": phone_number_id}, + "contacts": [{"wa_id": "15551234567", "profile": {"name": "Alice"}}], + "messages": messages + if messages is not None + else [ + { + "id": "wamid.incoming", + "from": "15551234567", + "type": "text", + "text": {"body": "hello"}, + } + ], + }, + } + ], + } + ], + } + + +class WhatsAppSession: + def __init__(self, receipt): + self.receipt, self.closed = receipt, False + self.status = 200 + + def post(self, *args, **kwargs): + self.value = self.receipt.get(*args, **kwargs) + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + pass + + async def json(self): + return {"messages": [{"id": self.value}]} + + async def close(self): + self.closed = True From 01eaba30c305d483cddf591d9c8998af47cb6a7a Mon Sep 17 00:00:00 2001 From: moe-berri Date: Mon, 12 Oct 2026 14:51:24 -0400 Subject: [PATCH 6/6] test: verify provider contracts and document shared delivery semantics --- README.md | 17 ++- docs/adding-a-channel.md | 7 + docs/architecture.md | 10 ++ docs/providers.md | 131 ++++++++++++++++++ tests/contracts/test_imports.py | 97 ++++++++++++++ tests/contracts/test_providers.py | 213 ++++++++++++++++++++++++++++++ 6 files changed, 473 insertions(+), 2 deletions(-) create mode 100644 docs/providers.md create mode 100644 tests/contracts/test_imports.py create mode 100644 tests/contracts/test_providers.py diff --git a/README.md b/README.md index da06ee2..65077e7 100644 --- a/README.md +++ b/README.md @@ -3,8 +3,21 @@ Connect chat platforms to any Python agent. AgentChat handles message delivery, conversation history, deduplication, and replies; your application owns the agent. -Slack is currently implemented. Each messaging platform lives in its own channel -adapter, so adding another platform does not require changing the message loop. +Slack, Discord, Microsoft Teams, iMessage through BlueBubbles, and WhatsApp Business +Cloud API have channel adapters. Each provider lives in its own package and uses +the same message loop. + +| Connect | Setup and echo example | +| --- | --- | +| Slack | [Slack guide](docs/slack.md) · [echo](examples/slack_echo.py) | +| Discord | [Discord guide](docs/discord.md) · [echo](examples/discord_echo.py) | +| Microsoft Teams | [Teams guide](docs/teams.md) · [echo](examples/teams_echo.py) | +| iMessage | [BlueBubbles guide](docs/imessage.md) · [echo](examples/imessage_echo.py) | +| WhatsApp | [Cloud API guide](docs/whatsapp.md) · [echo](examples/whatsapp_echo.py) | + +The four new adapters support receiving messages and confirmed text replies. +See [shared usage, capabilities, and delivery limits](docs/providers.md) for +multiple providers or an existing agent runner. ## Quickstart diff --git a/docs/adding-a-channel.md b/docs/adding-a-channel.md index d2ca2ba..09c01d8 100644 --- a/docs/adding-a-channel.md +++ b/docs/adding-a-channel.md @@ -76,6 +76,13 @@ deduplication and failure propagation. Add a harness fixture for your provider i assertions against it. Each harness adapts native event injection and outbound transport inspection; the contract tests remain provider-independent. +The new gateway/HTTP adapters also run the parametrized suite in +[`test_providers.py`](../tests/contracts/test_providers.py), covering receipts, +dispatch/deduplication, absent capabilities, queue limits, and real lifecycle +cleanup. Its fixtures use helpers from `tests/support/.py`. Add a new +provider there when using the bounded background-dispatch helper; its errors +are logged asynchronously instead of escaping the event callback. + Also cover provider-specific workspace checks, DM/thread routing, subscription rules, malformed events, pagination, socket acknowledgement, queue limits and shutdown where applicable. Capability tests should exercise unsupported features, diff --git a/docs/architecture.md b/docs/architecture.md index 7502da2..84db455 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -74,6 +74,14 @@ Keep new provider implementations in their own packages. Add internal files when their responsibilities justify it; a small provider can start with `adapter.py`. See [adding a channel](adding-a-channel.md) for the extension contract. +Discord, Teams, iMessage, and WhatsApp follow this package boundary with +`adapter.py`, pure normalization in `events.py`, and text validation in +`rendering.py`. Discord and BlueBubbles have small REST transport modules too. +`channels/_delivery.py` shares bounded background dispatch and cleanup among +these adapters. It is a private composition helper, not a channel superclass; +provider authentication, acknowledgement, and overflow policy stay in each adapter. +The [provider guides](providers.md) describe their supported operations and limits. + ## Public imports and compatibility Both `from agentchat.channels import Slack` and @@ -85,6 +93,8 @@ The `channels` and `integrations` namespaces load concrete implementations only when requested. Importing the core does not import Slack, aiohttp or OpenAI Agents. Installation defaults are unchanged: Slack dependencies remain installed with the base package, while OpenAI Agents still uses the `openai` extra. +Discord, Teams, and iMessage use separate optional extras. WhatsApp uses aiohttp +from the base install. Requesting one provider does not import the other providers. The documented legacy helper modules forward to canonical implementations: diff --git a/docs/providers.md b/docs/providers.md new file mode 100644 index 0000000..dd75731 --- /dev/null +++ b/docs/providers.md @@ -0,0 +1,131 @@ +# Run messaging adapters + +Choose the setup guide for the account you want to connect. Each guide ends with +a runnable echo example; no agent framework or model key is needed for that check. + +| Provider | Install from checkout | Connection | Setup | +| --- | --- | --- | --- | +| Slack | `pip install -e .` | Socket Mode | [Slack](slack.md) | +| Discord | `pip install -e '.[discord]'` | Outbound gateway | [Discord](discord.md) | +| Teams | `pip install -e '.[teams]'` | Authenticated HTTPS endpoint | [Teams](teams.md) | +| iMessage | `pip install -e '.[imessage]'` | BlueBubbles Socket.IO + REST | [iMessage](imessage.md) | +| WhatsApp | `pip install -e '.[whatsapp]'` | Signed HTTPS webhook + Cloud API | [WhatsApp](whatsapp.md) | + +Slack's existing install and imports are unchanged. The `whatsapp` extra needs +no additional dependency because aiohttp is already installed. Provider packages +and agent-framework integrations load only when imported. Python 3.11+ is required. + +## Connect your runner + +After completing the provider setups, construct just the channels you need: + +```python +import asyncio + +from agentchat import AgentChat +from agentchat.channels import Discord, IMessage, Teams, WhatsApp + + +async def main(): + app = AgentChat(channels=[ + Discord.from_env(), Teams.from_env(), IMessage.from_env(), WhatsApp.from_env(), + ]) + + @app.on_message + async def respond(context): + # Authorize context.sender before fetching context or invoking agent tools. + return context.message.text or "Received your attachment." + + try: + await app.run() + finally: + await app.close() + + +asyncio.run(main()) +``` + +Install several extras together, for example `pip install -e '.[discord,teams,imessage,whatsapp]'`. +The namespaces in `Message.id` and `conversation_id` keep provider/account histories +separate. Replacing the echo handler with an agent does not change transport setup. + +If your host already owns deduplication, ordering, and history, bind it directly: + +```python +channel = Discord.from_env() + + +async def receive(channel, message): + # Authorize the sender, claim the ID, and serialize this conversation in your host. + await channel.reply(message, "Received.") + + +async def main(): + channel.bind(receive) + try: + await channel.run() + finally: + await channel.close() +``` + +Call `asyncio.run(main())` in your executable. Retain the full source `Message` +when handing work to a worker: its metadata contains provider-specific reply routing. +Resolve that source from your trusted store rather than accepting destinations +or account IDs from an agent or browser input. + +## Capabilities + +These four new adapters implement the minimal `Channel` contract: receive, +lifecycle, and confirmed text replies. They normalize sender identity and +incoming attachment references. They do not download those attachments. + +History uses the application's local `State`; default `MemoryState` is lost on +restart. Native history, working indicators, rich replies, file transfers, and +cross-surface mirroring remain optional capabilities. On the new adapters, +`set_status()` returns `False`; unsupported rich/file/mirror operations raise +`TypeError`. Slack retains its existing capabilities. Applications choose text +or link fallbacks explicitly rather than passing Slack Block Kit to another provider. + +Long responses raise a validation error at the provider's documented limit; +there is no silent truncation or automatic multi-message send. If you split an +answer, track each `context.reply()` independently because earlier chunks may +have succeeded before a later send fails. + +## Delivery and shutdown + +Each adapter admits at most `max_pending_events=64` in-process handler tasks. +`AgentChat` deduplicates accepted message IDs and serializes handlers by +conversation using the configured `State`. + +- Teams returns 202 after admission; WhatsApp returns 200. A full HTTP receiver + returns 503 before admission so the provider can apply its retry policy. + WhatsApp batches are admitted atomically. +- Discord and BlueBubbles callbacks cannot ask the gateway to replay an event. + At capacity, the adapter logs a warning and drops the excess event. Reconnects + do not provide durable catch-up. Size the limit/host for the expected workload. +- `close()` stops admission, drains handlers for up to `shutdown_timeout=20` + seconds, then cancels unfinished handlers and releases clients/listeners. + `run()` cleans up on cancellation or connection/startup failure. Instances are + intended for a single run; create a new instance after closing one. + +Accepted work is not a durable queue. A crash after webhook acknowledgement can +lose a pending handler; a failed handler is logged and not automatically replayed. +Use application-owned durable ingestion/state/outboxes for stronger guarantees. +Persisting only a claim before work is durable can suppress a provider retry. + +Replies make a single send attempt and return a normalized assistant message only +after obtaining a provider receipt. The core appends that confirmed result to +history. A receipt means provider acceptance, not that a human received/read it. +Timeouts may occur after a provider accepts a message; avoid blindly retrying +uncertain writes. Rate-limit scheduling and delivery reconciliation belong to the +host. Injected clients must preserve these receipt/retry semantics. + +## Validate a deployment + +The test suite covers provider event fixtures, Microsoft/Discord SDK behavior, +authentication rejection, local HTTP and Socket.IO round trips, import isolation, +receipts, failure handling, and cleanup. It uses no live platform credentials. +Run the guide's live echo check for each configured account before enabling your +agent, including a channel/thread check where relevant. Successful local tests +do not verify provider installation, tenant permissions, phone registration, or +macOS automation permissions. diff --git a/tests/contracts/test_imports.py b/tests/contracts/test_imports.py new file mode 100644 index 0000000..2ca3bef --- /dev/null +++ b/tests/contracts/test_imports.py @@ -0,0 +1,97 @@ +import subprocess +import sys + +import pytest + + +def run_script(code): + result = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, check=False + ) + assert result.returncode == 0, result.stderr + + +BLOCK_IMPORTS = """ +import sys +from importlib.abc import MetaPathFinder +class Block(MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname.split('.')[0] in {'slack_sdk', 'agents', 'openai', 'discord', + 'microsoft_agents', 'socketio'}: + raise ModuleNotFoundError('blocked optional dependency', name=fullname) +sys.meta_path.insert(0, Block()) +""" + + +def test_core_and_namespaces_do_not_import_provider_or_framework_sdks(): + run_script( + BLOCK_IMPORTS + + """ +from agentchat import AgentChat, Message, MessageContext, Sender +from agentchat.channels import Channel, ThreadHistoryChannel +import agentchat.integrations +from agentchat.channels.whatsapp import WhatsApp +assert 'agentchat.channels.slack' not in sys.modules +assert 'agentchat.integrations.openai_agents' not in sys.modules +""" + ) + + +@pytest.mark.parametrize( + "provider,extra", + [ + ("Discord", "discord"), + ("Teams", "teams"), + ("IMessage", "imessage"), + ], +) +def test_optional_provider_has_actionable_import_error(provider, extra): + run_script( + BLOCK_IMPORTS + + f""" +try: + from agentchat.channels import {provider} +except ImportError as error: + assert 'agentchat[{extra}]' in str(error), str(error) +else: + raise AssertionError('provider import should require optional SDK') +""" + ) + + +def test_openai_helpers_remain_lazy_and_actionable(): + run_script( + BLOCK_IMPORTS + + """ +try: + from agentchat.integrations import should_reply +except ImportError as error: + assert 'agentchat[openai]' in str(error), str(error) +else: + raise AssertionError('integration import should require optional SDK') +""" + ) + + +@pytest.mark.parametrize("order", ["explicit_first", "namespace_first"]) +def test_export_identity_in_both_import_orders(order): + explicit = """ +from agentchat.channels.slack import Slack as SlackDirect, SlackUser as SlackUserDirect +from agentchat.channels.discord import Discord as DiscordDirect +from agentchat.channels.teams import Teams as TeamsDirect +from agentchat.channels.imessage import IMessage as IMessageDirect +from agentchat.channels.whatsapp import WhatsApp as WhatsAppDirect +from agentchat.integrations.openai_agents import should_reply as should_reply_direct +""" + namespace = """ +from agentchat.channels import Slack, SlackUser, Discord, Teams, IMessage, WhatsApp +from agentchat.integrations import should_reply +""" + checks = """ +assert Slack is SlackDirect and SlackUser is SlackUserDirect +assert Discord is DiscordDirect and Teams is TeamsDirect +assert IMessage is IMessageDirect and WhatsApp is WhatsAppDirect +assert should_reply is should_reply_direct +""" + code = explicit + namespace if order == "explicit_first" else namespace + explicit + run_script(code + checks) diff --git a/tests/contracts/test_providers.py b/tests/contracts/test_providers.py new file mode 100644 index 0000000..6b7f2db --- /dev/null +++ b/tests/contracts/test_providers.py @@ -0,0 +1,213 @@ +import asyncio +from dataclasses import replace +from types import SimpleNamespace + +import pytest +from aiohttp import web + +from agentchat import AgentChat, RichReply, UploadFile +from agentchat.channels import Discord, IMessage, Teams, WhatsApp +from agentchat.channels.discord.events import normalize as discord_message +from agentchat.channels.imessage.events import normalize as imessage_message +from agentchat.channels.teams.events import normalize as teams_message +from agentchat.channels.whatsapp.events import normalize as whatsapp_messages +from tests.support.discord import DiscordClient, discord_event +from tests.support.imessage import BubblesClient, SocketClient, imessage_event +from tests.support.receipts import Receipt +from tests.support.teams import APP_ID, TENANT_ID, teams_adapter, teams_event +from tests.support.whatsapp import WhatsAppSession, whatsapp_event + + +@pytest.fixture(params=["discord", "teams", "imessage", "whatsapp"]) +def harness(request): + receipt = Receipt(101 if request.param == "discord" else "sent") + options = {"max_pending_events": 2, "shutdown_timeout": 0.01} + if request.param == "discord": + client = DiscordClient(receipt) + channel = Discord(bot_token="fake", client=client, send_client=client, **options) + source = discord_message(discord_event(), bot_id=99) + elif request.param == "teams": + adapter, _, _ = teams_adapter(receipt) + channel = Teams( + app_id=APP_ID, + tenant_id=TENANT_ID, + app_password="fake", + adapter=adapter, + port=0, + **options, + ) + source = teams_message(teams_event(), app_id=APP_ID, tenant_id=TENANT_ID) + elif request.param == "imessage": + channel = IMessage( + server_url="http://localhost:1234", + password="fake", + instance_id="test", + client=BubblesClient(receipt), + socket_client=SocketClient(), + **options, + ) + source = imessage_message(imessage_event(), instance_id="test") + else: + channel = WhatsApp( + phone_number_id="123", + access_token="fake", + app_secret="secret", + verify_token="verify", + port=0, + **options, + ) + channel._session = WhatsAppSession(receipt) + source = whatsapp_messages(whatsapp_event(), phone_number_id="123")[0] + return SimpleNamespace(channel=channel, source=source, receipt=receipt) + + +@pytest.mark.asyncio +async def test_confirmed_reply_preserves_conversation_and_dedupes_inbound(harness): + app = AgentChat(channels=[harness.channel]) + seen = [] + + @app.on_message + async def respond(context): + seen.append(context.message) + return "confirmed answer" + + try: + await app._receive(harness.channel, harness.source) + await app._receive(harness.channel, harness.source) + history = await app.state.history(harness.source.conversation_id) + assert seen == [harness.source] + assert len(harness.receipt.calls) == 1 + assert len(history) == 2 + reply = history[1] + assert reply.role == "assistant" + assert reply.channel == harness.source.channel + assert reply.thread_id == harness.source.thread_id + assert reply.conversation_id == harness.source.conversation_id + assert reply.id != harness.source.id + assert reply.text == "confirmed answer" + finally: + await app.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", ["timeout", "missing_receipt"]) +async def test_failed_or_uncertain_send_never_appends_or_retries(harness, failure): + app = AgentChat(channels=[harness.channel]) + if failure == "timeout": + harness.receipt.error = TimeoutError("uncertain") + else: + harness.receipt.value = None + try: + with pytest.raises((TimeoutError, RuntimeError)): + await app.reply(harness.channel, harness.source, "answer") + assert len(harness.receipt.calls) == 1 + assert await app.state.history(harness.source.conversation_id) == () + finally: + await app.close() + + +@pytest.mark.asyncio +async def test_rejects_wrong_provider_and_empty_text_before_send(harness): + try: + with pytest.raises(ValueError): + await harness.channel.reply(replace(harness.source, channel="wrong"), "answer") + with pytest.raises(ValueError): + await harness.channel.reply(harness.source, " ") + assert not harness.receipt.calls + finally: + await harness.channel.close() + + +@pytest.mark.asyncio +async def test_optional_capability_fallbacks(harness): + app = AgentChat(channels=[harness.channel]) + try: + assert await app.set_status(harness.channel, harness.source, "working") is False + with pytest.raises(TypeError, match="rich replies"): + await app.reply_rich(harness.channel, harness.source, RichReply(text="answer")) + with pytest.raises(TypeError, match="file uploads"): + await app.upload_files(harness.channel, harness.source, [UploadFile("a.txt", b"a")]) + with pytest.raises(TypeError, match="mirroring"): + await app.mirror(harness.channel, harness.source, harness.source) + finally: + await app.close() + + +@pytest.mark.asyncio +async def test_run_requires_bind(harness): + with pytest.raises(RuntimeError, match="Bind"): + await harness.channel.run() + await harness.channel.close() + + +@pytest.mark.asyncio +async def test_close_rejects_new_sends_but_drains_accepted_replies(harness): + harness.channel._delivery.shutdown_timeout = 1 + + async def receive(channel, message): + await channel.reply(message, "last answer") + + harness.channel.bind(receive) + assert harness.channel._delivery.submit(harness.channel, harness.source) + await harness.channel.close() + assert len(harness.receipt.calls) == 1 + with pytest.raises(RuntimeError, match="closed"): + await harness.channel.reply(harness.source, "too late") + assert len(harness.receipt.calls) == 1 + + +@pytest.mark.asyncio +async def test_direct_binding_backpressure_and_shutdown_cancellation(harness): + started, cancelled = asyncio.Event(), asyncio.Event() + + async def receive(channel, message): + assert channel is harness.channel + started.set() + try: + await asyncio.Event().wait() + finally: + cancelled.set() + + harness.channel.bind(receive) + delivery = harness.channel._delivery + assert delivery.submit(harness.channel, harness.source) + assert delivery.submit(harness.channel, replace(harness.source, id="second")) + assert not delivery.submit(harness.channel, replace(harness.source, id="overflow")) + await asyncio.wait_for(started.wait(), timeout=1) + await harness.channel.close() + await harness.channel.close() + assert cancelled.is_set() + assert not delivery.tasks + assert not delivery.submit(harness.channel, harness.source) + + +@pytest.mark.asyncio +async def test_run_and_close_release_resources(harness, monkeypatch): + async def receive(channel, message): + pass + + harness.channel.bind(receive) + started = asyncio.Event() + start_site = web.TCPSite.start + + async def start(site): + await start_site(site) + started.set() + + monkeypatch.setattr(web.TCPSite, "start", start) + running = asyncio.create_task(harness.channel.run()) + try: + async with asyncio.timeout(2): + if harness.channel.name == "discord": + await harness.channel._client.started.wait() + elif harness.channel.name == "imessage": + await harness.channel._socket.started.wait() + else: + await started.wait() + await harness.channel.close() + await asyncio.wait_for(running, timeout=2) + assert harness.channel._delivery.closed + finally: + running.cancel() + await asyncio.gather(running, return_exceptions=True) + await harness.channel.close()