From 2ddcf1cc55e8d8a28732537aef2824bf53d39c05 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 11:01:49 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E6=B7=BB=E5=8A=A0=20Intercom?= =?UTF-8?q?=20=E6=B8=A0=E9=81=93=E6=89=A9=E5=B1=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 Intercom 渠道扩展，支持在 Yuxi 平台中集成 Intercom 客服渠道。 包含以下功能模块： - client: Intercom API 客户端封装 - config: 渠道配置管理 - gateway: SSE/WebSocket 网关接入 - webhook: Webhook 事件处理 - streaming: 流式消息处理 - format: 消息格式转换 - outbound: 外发消息管理 - handoff: 转人工会话切换 - pairing: 用户配对与绑定 - security: 安全签名校验 - dedupe: 消息去重 - monitor: 渠道状态监控 - status: 会话状态管理 - session_guard: 会话防护 - types: 类型定义 --- .../channel/extensions/intercom/__init__.py | 490 +++++++++++++++ .../channel/extensions/intercom/client.py | 579 ++++++++++++++++++ .../channel/extensions/intercom/config.py | 192 ++++++ .../channel/extensions/intercom/constants.py | 72 +++ .../channel/extensions/intercom/dedupe.py | 62 ++ .../channel/extensions/intercom/errors.py | 51 ++ .../channel/extensions/intercom/format.py | 106 ++++ .../channel/extensions/intercom/gateway.py | 128 ++++ .../channel/extensions/intercom/handoff.py | 54 ++ .../channel/extensions/intercom/monitor.py | 54 ++ .../channel/extensions/intercom/outbound.py | 371 +++++++++++ .../channel/extensions/intercom/pairing.py | 43 ++ .../channel/extensions/intercom/plugin.json | 28 + .../channel/extensions/intercom/security.py | 66 ++ .../extensions/intercom/session_guard.py | 12 + .../channel/extensions/intercom/status.py | 187 ++++++ .../channel/extensions/intercom/streaming.py | 67 ++ .../yuxi/channel/extensions/intercom/types.py | 71 +++ .../channel/extensions/intercom/webhook.py | 265 ++++++++ 19 files changed, 2898 insertions(+) create mode 100644 backend/package/yuxi/channel/extensions/intercom/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/client.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/config.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/constants.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/dedupe.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/errors.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/format.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/gateway.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/handoff.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/monitor.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/outbound.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/pairing.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/plugin.json create mode 100644 backend/package/yuxi/channel/extensions/intercom/security.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/session_guard.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/status.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/streaming.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/types.py create mode 100644 backend/package/yuxi/channel/extensions/intercom/webhook.py diff --git a/backend/package/yuxi/channel/extensions/intercom/__init__.py b/backend/package/yuxi/channel/extensions/intercom/__init__.py new file mode 100644 index 00000000..168a64e0 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/intercom/__init__.py @@ -0,0 +1,490 @@ +import logging + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.intercom.config import IntercomConfigAdapter +from yuxi.channel.extensions.intercom.errors import IntercomError +from yuxi.channel.extensions.intercom.format import get_agent_prompt_rules +from yuxi.channel.extensions.intercom.gateway import IntercomGateway +from yuxi.channel.extensions.intercom.monitor import IntercomMonitor +from yuxi.channel.extensions.intercom.outbound import IntercomOutbound +from yuxi.channel.extensions.intercom.pairing import IntercomPairing +from yuxi.channel.extensions.intercom.security import IntercomSecurity +from yuxi.channel.extensions.intercom.status import IntercomStatus +from yuxi.channel.extensions.intercom.streaming import IntercomStreaming +from yuxi.channel.plugins.registry import ChannelPluginRegistry +from yuxi.channel.protocols import ClassifiedError, ErrorSeverity + +logger = logging.getLogger(__name__) + + +class IntercomPlugin(BaseChannelPlugin): + id = "intercom" + name = "Intercom" + order = 65 + label = "Intercom" + aliases = ["intercom-messenger"] + + def __init__(self): + self._config = IntercomConfigAdapter() + self._gateway = IntercomGateway() + self._outbound = IntercomOutbound(gateway=self._gateway) + self._monitor = IntercomMonitor() + self._streaming = IntercomStreaming() + self._security = IntercomSecurity() + self._pairing = IntercomPairing() + self._status = IntercomStatus() + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct"], + message_types=["text", "image", "file"], + interactions=["button"], + reactions=False, + typing_indicator=False, + threads=False, + edit=False, + unsend=False, + reply=True, + media=True, + native_commands=False, + polls=False, + group_management=False, + streaming=True, + streaming_mode="block", + block_streaming=True, + tts=None, + ) + + # ── Config 委托 ────────────────────────────────────── + + def list_account_ids(self, config: dict) -> list[str]: + return self._config.list_account_ids(config) + + async def resolve_account(self, account_id: str, config: dict | None = None) -> dict: + return await self._config.resolve_account(account_id, config) + + def is_configured(self, account: dict) -> bool: + return self._config.is_configured(account) + + def is_enabled(self, account: dict, config: dict) -> bool: + return self._config.is_enabled(account, config) + + def disabled_reason(self, account: dict, config: dict) -> str: + return self._config.disabled_reason(account, config) + + def unconfigured_reason(self, account: dict, config: dict) -> str: + return self._config.unconfigured_reason(account, config) + + def describe_account(self, account: dict, config: dict) -> dict: + return self._config.describe_account(account, config) + + def default_account_id(self, config: dict) -> str: + return self._config.default_account_id(config) + + def resolve_allow_from(self, config: dict, account_id: str | None = None) -> list[str] | None: + return self._config.resolve_allow_from(config, account_id) + + def format_allow_from(self, config: dict, account_id: str | None, allow_from: list[str | int]) -> list[str]: + return self._config.format_allow_from(config, account_id, allow_from) + + def has_configured_state(self, config: dict) -> bool: + return self._config.has_configured_state(config) + + def has_persisted_auth_state(self, config: dict) -> bool: + return self._config.has_persisted_auth_state(config) + + def config_schema(self) -> dict: + return self._config.config_schema() + + def apply_config_fixes(self, config: dict) -> dict: + return self._security.apply_config_fixes(config) + + def collect_warnings(self, config: dict, account_id: str | None = None, account: dict | None = None) -> list[str]: + return self._security.collect_warnings(config, account_id, account) + + # ── Gateway 委托 ───────────────────────────────────── + + @property + def delivery_mode(self): + return self._gateway.delivery_mode + + async def start(self, ctx) -> object: + result = await self._gateway.start(ctx) + + from yuxi.channel.extensions.intercom.webhook import set_gateway_instance + + set_gateway_instance(self._gateway) + + return result + + async def stop(self, ctx) -> None: + await self._gateway.stop(ctx) + + async def resolve_gateway_auth_bypass_paths(self, config: dict) -> list[str]: + return await self._gateway.resolve_gateway_auth_bypass_paths(config) + + # ── DedupeProtocol 委托 ────────────────────────────── + + def is_duplicate(self, key: str) -> bool: + return self._gateway.check_duplicate(key) + + def mark_seen(self, key: str) -> None: + self._gateway.mark_seen(key) + + def reset_dedupe(self) -> None: + self._gateway.reset_dedupe() + + # ── ErrorHandlingProtocol ───────────────────────────── + + def classify_error(self, error: BaseException) -> ClassifiedError: + if isinstance(error, IntercomError): + return ClassifiedError( + severity=ErrorSeverity.RETRYABLE if error.retryable else ErrorSeverity.FATAL, + error_message=str(error), + original_error=error, + ) + return ClassifiedError( + severity=ErrorSeverity.RETRYABLE, + error_message=str(error), + original_error=error, + ) + + def is_retryable(self, error: BaseException) -> bool: + if isinstance(error, IntercomError): + return error.retryable + return True + + # ── Outbound 委托 ──────────────────────────────────── + + @property + def text_chunk_limit(self) -> int | None: + return self._outbound.text_chunk_limit + + @property + def chunker_mode(self) -> str | None: + return self._outbound.chunker_mode + + @property + def poll_max_options(self) -> int | None: + return self._outbound.poll_max_options + + @property + def supports_poll_duration_seconds(self) -> bool: + return self._outbound.supports_poll_duration_seconds + + @property + def supports_anonymous_polls(self) -> bool: + return self._outbound.supports_anonymous_polls + + @property + def extract_markdown_images(self) -> bool: + return self._outbound.extract_markdown_images + + @property + def presentation_capabilities(self): + return self._outbound.presentation_capabilities + + @property + def delivery_capabilities(self): + return self._outbound.delivery_capabilities + + def chunker(self, text: str, limit: int, ctx: object | None = None) -> list[str]: + return self._outbound.chunker(text, limit, ctx) + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + await self._outbound.send_text( + target_id, + content, + reply_to_id=reply_to_id, + thread_id=thread_id, + account_id=account_id, + ) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + await self._outbound.send_media( + target_id, + media_url, + media_type, + reply_to_id=reply_to_id, + thread_id=thread_id, + account_id=account_id, + ) + + async def send_note( + self, + target_id: str, + content: str, + *, + account_id: str | None = None, + ) -> None: + await self._outbound.send_note(target_id, content, account_id=account_id) + + async def add_tags( + self, + target_id: str, + tags: list[str], + *, + account_id: str | None = None, + ) -> None: + await self._outbound.add_tags(target_id, tags, account_id=account_id) + + async def remove_tags( + self, + target_id: str, + tag_ids: list[str], + *, + account_id: str | None = None, + ) -> None: + await self._outbound.remove_tags(target_id, tag_ids, account_id=account_id) + + async def send_quick_replies( + self, + target_id: str, + content: str, + options: list[dict], + *, + account_id: str | None = None, + ) -> None: + await self._outbound.send_quick_replies(target_id, content, options, account_id=account_id) + + async def close_conversation(self, target_id: str, *, account_id: str | None = None) -> bool: + return await self._outbound.close_conversation(target_id, account_id=account_id) + + async def open_conversation(self, target_id: str, *, account_id: str | None = None) -> bool: + return await self._outbound.open_conversation(target_id, account_id=account_id) + + async def create_conversation( + self, + contact_id: str, + body: str, + *, + account_id: str | None = None, + ) -> dict | None: + return await self._outbound.create_conversation(contact_id, body, account_id=account_id) + + async def convert_to_ticket(self, conversation_id: str, *, account_id: str | None = None) -> dict | None: + return await self._outbound.convert_to_ticket(conversation_id, account_id=account_id) + + async def send_payload(self, ctx: object) -> object: + return await self._outbound.send_payload(ctx) + + async def send_poll(self, ctx: object) -> object: + return await self._outbound.send_poll(ctx) + + def sanitize_text(self, text: str, payload: object) -> str: + return self._outbound.sanitize_text(text, payload) + + def should_skip_plain_text_sanitization(self, payload: object) -> bool: + return self._outbound.should_skip_plain_text_sanitization(payload) + + def normalize_payload(self, payload, config, account_id=None): + return self._outbound.normalize_payload(payload, config, account_id) + + def resolve_effective_text_chunk_limit(self, config, account_id=None, fallback_limit=None): + return self._outbound.resolve_effective_text_chunk_limit(config, account_id, fallback_limit) + + # ── Streaming 委托 ─────────────────────────────────── + + @property + def streaming_mode(self) -> str: + return self._streaming.streaming_mode + + @property + def preview_stream_throttle_ms(self) -> int: + return self._streaming.preview_stream_throttle_ms + + @property + def preview_min_initial_chars(self) -> int: + return self._streaming.preview_min_initial_chars + + @property + def block_streaming_enabled(self) -> bool: + return self._streaming.block_streaming_enabled + + @property + def block_streaming_break(self) -> str: + return self._streaming.block_streaming_break + + @property + def block_streaming_chunk_min_chars(self) -> int: + return self._streaming.block_streaming_chunk_min_chars + + @property + def block_streaming_chunk_max_chars(self) -> int: + return self._streaming.block_streaming_chunk_max_chars + + @property + def block_streaming_chunk_break_preference(self) -> str: + return self._streaming.block_streaming_chunk_break_preference + + @property + def block_streaming_coalesce_defaults(self): + return self._streaming.block_streaming_coalesce_defaults + + def create_draft_stream_session(self, target_id: str) -> object: + return self._streaming.create_draft_stream_session(target_id) + + def create_block_chunker(self) -> object: + return self._streaming.create_block_chunker() + + # ── Security 委托 ──────────────────────────────────── + + def resolve_dm_policy(self) -> dict: + return self._security.resolve_dm_policy() + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + return await self._security.check_allowlist(peer_id, channel_type) + + def resolve_dm_policy_for_account(self, account: dict) -> dict: + return self._security.resolve_dm_policy_for_account(account) + + def collect_audit_findings( + self, + config, + account_id=None, + account=None, + source_config=None, + ordered_account_ids=None, + has_explicit_account_path=False, + ) -> list[dict]: + return self._security.collect_audit_findings( + config, + account_id, + account, + source_config, + ordered_account_ids, + has_explicit_account_path, + ) + + # ── Pairing 委托 ───────────────────────────────────── + + @property + def id_label(self) -> str: + return self._pairing.id_label + + async def generate_code(self, peer_id: str) -> str: + return await self._pairing.generate_code(peer_id) + + async def verify_code(self, peer_id: str, code: str) -> bool: + return await self._pairing.verify_code(peer_id, code) + + def normalize_allow_entry(self, entry: str) -> str: + return self._pairing.normalize_allow_entry(entry) + + async def notify_approval(self, config: dict, peer_id: str, account_id: str | None = None) -> None: + await self._pairing.notify_approval(config, peer_id, account_id) + + # ── Status 委托 ────────────────────────────────────── + + async def probe(self, account: dict | None = None) -> bool: + return await self._status.probe(account) + + def build_summary(self, snapshot: object) -> dict: + return self._status.build_summary(snapshot) + + def build_channel_summary(self, account: dict, config: dict, default_account_id: str, snapshot) -> dict: + return self._status.build_channel_summary(account, config, default_account_id, snapshot) + + def build_account_snapshot(self, account: dict, config: dict, runtime=None, probe=None, audit=None): + return self._status.build_account_snapshot(account, config, runtime, probe, audit) + + def format_capabilities_probe(self, probe: object) -> list[dict]: + return self._status.format_capabilities_probe(probe) + + async def audit_account(self, account: dict, timeout_ms: int, config: dict, probe: object | None = None) -> object: + return await self._status.audit_account(account, timeout_ms, config, probe) + + async def build_capabilities_diagnostics( + self, + account: dict, + timeout_ms: int, + config: dict, + probe=None, + audit=None, + target=None, + ) -> dict | None: + return await self._status.build_capabilities_diagnostics( + account, + timeout_ms, + config, + probe, + audit, + target, + ) + + def log_self_id(self, account: dict, config: dict, runtime: object, include_channel_prefix: bool = False) -> None: + self._status.log_self_id(account, config, runtime, include_channel_prefix) + + def resolve_account_state(self, account: dict, config: dict, configured: bool, enabled: bool) -> str: + return self._status.resolve_account_state(account, config, configured, enabled) + + def collect_status_issues(self, accounts: list) -> list: + return self._status.collect_status_issues(accounts) + + # ── Agent Prompt ────────────────────────────────────── + + @property + def channel_format_instructions(self) -> str | None: + return "\n".join(get_agent_prompt_rules()) + + def build_system_prompt(self, context) -> str | None: + lines = get_agent_prompt_rules() + return "\n".join(lines) + + # ── Session ─────────────────────────────────────────── + + def resolve_session(self, msg) -> object: + from yuxi.channel.protocols import SessionResolution + + conversation_id = "" + if hasattr(msg, "group") and msg.group and msg.group.id: + conversation_id = msg.group.id + + return SessionResolution(kind="direct", conversation_id=conversation_id) + + # ── Lifecycle ───────────────────────────────────────── + + async def run_startup_maintenance(self, cfg: dict) -> None: + channel_cfg = cfg.get("channels", {}).get("intercom", {}) + if not channel_cfg: + logger.info("Intercom not configured, skipping startup maintenance") + return + + account_ids = self._config.list_account_ids(cfg) + for aid in account_ids: + account = await self._config.resolve_account(aid, cfg) + if not self._config.is_configured(account): + logger.warning("Intercom account [%s] is not configured", aid) + continue + + warnings = self._security.collect_warnings(cfg, aid, account) + for w in warnings: + logger.warning("Intercom [%s]: %s", aid, w) + + if not account.get("webhook_secret"): + logger.warning( + "Intercom [%s]: webhookSecret not configured. Webhook signature verification will be skipped.", + aid, + ) + + +intercom_plugin = ChannelPluginRegistry.register(IntercomPlugin()) diff --git a/backend/package/yuxi/channel/extensions/intercom/client.py b/backend/package/yuxi/channel/extensions/intercom/client.py new file mode 100644 index 00000000..a5143741 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/intercom/client.py @@ -0,0 +1,579 @@ +import logging + +import httpx + +from yuxi.channel.extensions.intercom.constants import ( + DEFAULT_DATACENTER, + INTERCOM_API_BASES, + INTERCOM_API_VERSION, + INTERCOM_DEFAULT_TIMEOUT, +) +from yuxi.channel.extensions.intercom.errors import IntercomError, IntercomErrorCode + +logger = logging.getLogger(__name__) + + +class IntercomClient: + def __init__( + self, + access_token: str, + *, + datacenter: str = DEFAULT_DATACENTER, + timeout: float = INTERCOM_DEFAULT_TIMEOUT, + ): + self._access_token = access_token + base_url = INTERCOM_API_BASES.get(datacenter, INTERCOM_API_BASES[DEFAULT_DATACENTER]) + self._client = httpx.AsyncClient( + base_url=base_url, + headers={ + "Authorization": f"Bearer {access_token}", + "Accept": "application/json", + "Intercom-Version": INTERCOM_API_VERSION, + }, + timeout=httpx.Timeout(timeout), + ) + + async def get_me(self) -> dict: + resp = await self._request("GET", "/me") + return resp + + async def reply_conversation( + self, + conversation_id: str, + body: str, + *, + message_type: str = "comment", + reply_type: str = "admin", + admin_id: str | None = None, + attachment_urls: list[str] | None = None, + reply_options: dict | None = None, + ) -> dict: + payload: dict = { + "message_type": message_type, + "type": reply_type, + "body": body, + } + if admin_id: + payload["admin_id"] = admin_id + if attachment_urls: + payload["attachment_urls"] = attachment_urls + if reply_options: + payload["reply_options"] = reply_options + + resp = await self._request( + "POST", + f"/conversations/{conversation_id}/reply", + json=payload, + ) + return resp + + async def get_conversation(self, conversation_id: str) -> dict: + resp = await self._request("GET", f"/conversations/{conversation_id}") + return resp + + async def get_conversation_parts(self, conversation_id: str) -> list[dict]: + resp = await self._request( + "GET", + f"/conversations/{conversation_id}", + params={"display_as": "plaintext"}, + ) + return resp.get("conversation_parts", {}).get("conversation_parts", []) + + async def manage_conversation( + self, + conversation_id: str, + action: str, + admin_id: str | None = None, + ) -> dict: + payload: dict = {"type": "admin"} + if admin_id: + payload["admin_id"] = admin_id + resp = await self._request( + "POST", + f"/conversations/{conversation_id}/manage", + json=payload, + params={"action": action}, + ) + return resp + + async def create_conversation( + self, + contact_id: str, + body: str, + *, + admin_id: str | None = None, + message_type: str = "comment", + attachment_urls: list[str] | None = None, + ) -> dict: + payload: dict = { + "from": {"type": "contact", "id": contact_id}, + "body": body, + } + if admin_id: + payload["admin_id"] = admin_id + if message_type: + payload["message_type"] = message_type + if attachment_urls: + payload["attachment_urls"] = attachment_urls + resp = await self._request("POST", "/conversations", json=payload) + return resp + + async def search_conversations( + self, + query: dict, + *, + page: int = 1, + per_page: int = 20, + ) -> dict: + payload = { + "query": query, + "pagination": {"page": page, "per_page": min(per_page, 150)}, + } + resp = await self._request("POST", "/conversations/search", json=payload) + return resp + + async def get_contact(self, contact_id: str) -> dict: + resp = await self._request("GET", f"/contacts/{contact_id}") + return resp + + async def create_contact(self, email: str, *, name: str | None = None, **attrs) -> dict: + payload: dict = {"email": email} + if name: + payload["name"] = name + payload.update(attrs) + resp = await self._request("POST", "/contacts", json=payload) + return resp + + async def update_contact(self, contact_id: str, **attrs) -> dict: + resp = await self._request("PUT", f"/contacts/{contact_id}", json=attrs) + return resp + + async def delete_contact(self, contact_id: str) -> dict: + resp = await self._request("DELETE", f"/contacts/{contact_id}") + return resp + + async def list_contacts(self, *, page: int = 1, per_page: int = 20, email: str | None = None) -> dict: + params: dict = {"page": page, "per_page": min(per_page, 150)} + if email: + params["email"] = email + resp = await self._request("GET", "/contacts", params=params) + return resp + + async def search_contacts( + self, + query: dict, + *, + page: int = 1, + per_page: int = 20, + ) -> dict: + payload = { + "query": query, + "pagination": {"page": page, "per_page": min(per_page, 150)}, + } + resp = await self._request("POST", "/contacts/search", json=payload) + return resp + + async def archive_contact(self, contact_id: str) -> dict: + resp = await self._request("POST", f"/contacts/{contact_id}/archive") + return resp + + async def add_conversation_tags(self, conversation_id: str, tags: list[str]) -> dict: + payload = {"tags": [{"id": tag} for tag in tags]} + resp = await self._request( + "POST", + f"/conversations/{conversation_id}/tags", + json=payload, + ) + return resp + + async def remove_conversation_tag(self, conversation_id: str, tag_id: str) -> dict: + resp = await self._request( + "DELETE", + f"/conversations/{conversation_id}/tags/{tag_id}", + ) + return resp + + async def list_tags(self) -> list[dict]: + resp = await self._request("GET", "/tags") + return resp.get("data", []) + + async def update_conversation(self, conversation_id: str, **kwargs) -> dict: + resp = await self._request( + "PUT", + f"/conversations/{conversation_id}", + json=kwargs, + ) + return resp + + async def snooze_conversation(self, conversation_id: str, snooze_until: int) -> dict: + return await self.update_conversation(conversation_id, snoozed_until=snooze_until) + + async def assign_conversation(self, conversation_id: str, admin_id: str) -> dict: + return await self.update_conversation(conversation_id, assignee_id=admin_id, assignee_type="admin") + + async def delete_conversation_part(self, conversation_id: str, part_id: str) -> dict: + resp = await self._request( + "DELETE", + f"/conversations/{conversation_id}/parts/{part_id}", + ) + return resp + + async def create_ticket( + self, + subject: str, + message_body: str, + contact_id: str, + *, + admin_id: str | None = None, + ticket_type_id: str | None = None, + ticket_attributes: dict | None = None, + ) -> dict: + payload: dict = { + "contacts": [{"id": contact_id}], + "subject": subject, + "ticket_type": {"body": message_body}, + } + if admin_id: + payload["admin_assignee_id"] = admin_id + if ticket_type_id: + payload["ticket_type_id"] = ticket_type_id + if ticket_attributes: + payload["ticket_attributes"] = ticket_attributes + resp = await self._request("POST", "/tickets", json=payload) + return resp + + async def reply_ticket( + self, + ticket_id: str, + body: str, + *, + admin_id: str | None = None, + attachment_urls: list[str] | None = None, + ) -> dict: + payload: dict = { + "type": "admin", + "body": body, + } + if admin_id: + payload["admin_id"] = admin_id + if attachment_urls: + payload["attachment_urls"] = attachment_urls + resp = await self._request("POST", f"/tickets/{ticket_id}/reply", json=payload) + return resp + + async def get_ticket(self, ticket_id: str) -> dict: + resp = await self._request("GET", f"/tickets/{ticket_id}") + return resp + + async def update_ticket(self, ticket_id: str, **kwargs) -> dict: + resp = await self._request("PUT", f"/tickets/{ticket_id}", json=kwargs) + return resp + + async def get_company(self, company_id: str) -> dict: + resp = await self._request("GET", f"/companies/{company_id}") + return resp + + async def create_company(self, name: str, *, company_id: str | None = None, **attrs) -> dict: + payload: dict = {"name": name} + if company_id: + payload["company_id"] = company_id + payload.update(attrs) + resp = await self._request("POST", "/companies", json=payload) + return resp + + async def update_company(self, company_id: str, **attrs) -> dict: + resp = await self._request("PUT", f"/companies/{company_id}", json=attrs) + return resp + + async def list_companies(self, *, page: int = 1, per_page: int = 20) -> dict: + resp = await self._request( + "GET", + "/companies", + params={"page": page, "per_page": min(per_page, 150)}, + ) + return resp + + async def search_companies( + self, + query: dict, + *, + page: int = 1, + per_page: int = 20, + ) -> dict: + payload = { + "query": query, + "pagination": {"page": page, "per_page": min(per_page, 150)}, + } + resp = await self._request("POST", "/companies/search", json=payload) + return resp + + async def attach_contact_to_company(self, contact_id: str, company_id: str) -> dict: + resp = await self._request( + "POST", + f"/contacts/{contact_id}/companies", + json={"id": company_id}, + ) + return resp + + async def detach_contact_from_company(self, contact_id: str, company_id: str) -> dict: + resp = await self._request( + "DELETE", + f"/contacts/{contact_id}/companies/{company_id}", + ) + return resp + + async def list_articles(self, *, page: int = 1, per_page: int = 20, state: str = "published") -> dict: + resp = await self._request( + "GET", + "/articles", + params={"page": page, "per_page": per_page, "state": state}, + ) + return resp + + async def get_article(self, article_id: str) -> dict: + resp = await self._request("GET", f"/articles/{article_id}") + return resp + + async def list_brands(self) -> list[dict]: + resp = await self._request("GET", "/brands") + return resp.get("data", []) + + async def get_brand(self, brand_id: str) -> dict: + resp = await self._request("GET", f"/brands/{brand_id}") + return resp + + async def list_conversations(self, *, page: int = 1, per_page: int = 20) -> dict: + resp = await self._request( + "GET", + "/conversations", + params={"page": page, "per_page": min(per_page, 150)}, + ) + return resp + + async def delete_conversation(self, conversation_id: str) -> dict: + resp = await self._request("DELETE", f"/conversations/{conversation_id}") + return resp + + async def redact_conversation_part(self, conversation_id: str, part_id: str) -> dict: + resp = await self._request( + "POST", + f"/conversations/{conversation_id}/parts/{part_id}/redact", + ) + return resp + + async def convert_conversation_to_ticket(self, conversation_id: str, *, admin_id: str | None = None) -> dict: + payload: dict = {} + if admin_id: + payload["admin_id"] = admin_id + resp = await self._request( + "POST", + f"/conversations/{conversation_id}/convert-to-ticket", + json=payload, + ) + return resp + + async def run_assignment_rules(self, conversation_id: str) -> dict: + resp = await self._request( + "POST", + f"/conversations/{conversation_id}/run_assignment_rules", + ) + return resp + + async def attach_contact_to_conversation(self, conversation_id: str, contact_id: str) -> dict: + resp = await self._request( + "POST", + f"/conversations/{conversation_id}/contacts", + json={"id": contact_id}, + ) + return resp + + async def detach_contact_from_conversation(self, conversation_id: str, contact_id: str) -> dict: + resp = await self._request( + "DELETE", + f"/conversations/{conversation_id}/contacts/{contact_id}", + ) + return resp + + async def list_tickets(self, *, page: int = 1, per_page: int = 20) -> dict: + resp = await self._request( + "GET", + "/tickets", + params={"page": page, "per_page": min(per_page, 150)}, + ) + return resp + + async def search_tickets( + self, + query: dict, + *, + page: int = 1, + per_page: int = 20, + ) -> dict: + payload = { + "query": query, + "pagination": {"page": page, "per_page": min(per_page, 150)}, + } + resp = await self._request("POST", "/tickets/search", json=payload) + return resp + + async def merge_contacts(self, primary_id: str, secondary_id: str) -> dict: + resp = await self._request( + "POST", + "/contacts/merge", + json={ + "primary_contact_id": primary_id, + "secondary_contact_id": secondary_id, + }, + ) + return resp + + async def add_contact_tags(self, contact_id: str, tags: list[str]) -> dict: + payload = {"tags": [{"id": tag} for tag in tags]} + resp = await self._request( + "POST", + f"/contacts/{contact_id}/tags", + json=payload, + ) + return resp + + async def remove_contact_tag(self, contact_id: str, tag_id: str) -> dict: + resp = await self._request( + "DELETE", + f"/contacts/{contact_id}/tags/{tag_id}", + ) + return resp + + async def get_contact_subscriptions(self, contact_id: str) -> dict: + resp = await self._request("GET", f"/contacts/{contact_id}/subscriptions") + return resp + + async def update_contact_subscriptions(self, contact_id: str, subscriptions: list[dict]) -> dict: + resp = await self._request( + "PUT", + f"/contacts/{contact_id}/subscriptions", + json={"subscriptions": subscriptions}, + ) + return resp + + async def send_email( + self, + to: list[str], + subject: str, + body_html: str, + *, + cc: list[str] | None = None, + bcc: list[str] | None = None, + ) -> dict: + payload: dict = { + "to": to, + "subject": subject, + "body": body_html, + } + if cc: + payload["cc"] = cc + if bcc: + payload["bcc"] = bcc + resp = await self._request("POST", "/emails", json=payload) + return resp + + async def send_message( + self, + contact_id: str, + body: str, + *, + admin_id: str | None = None, + ) -> dict: + payload: dict = { + "from": {"type": "admin", "id": admin_id} if admin_id else {"type": "admin"}, + "to": {"type": "contact", "id": contact_id}, + "body": body, + } + resp = await self._request("POST", "/messages", json=payload) + return resp + + async def close(self) -> None: + await self._client.aclose() + + async def _paginated_request(self, method: str, path: str, *, params: dict | None = None, **kwargs) -> list[dict]: + params = params or {} + results: list[dict] = [] + while True: + resp = await self._request(method, path, params=params, **kwargs) + data = resp.get("data", []) + results.extend(data) + pages_info = resp.get("pages", {}) + next_cursor = pages_info.get("next", {}).get("starting_after") + if not next_cursor: + break + params["starting_after"] = next_cursor + return results + + async def _request(self, method: str, path: str, **kwargs) -> dict: + try: + resp = await self._client.request(method, path, **kwargs) + except httpx.TimeoutException as e: + raise IntercomError( + IntercomErrorCode.NETWORK_ERROR, + f"Request timeout: {method} {path}", + retryable=True, + ) from e + except httpx.ConnectError as e: + raise IntercomError( + IntercomErrorCode.NETWORK_ERROR, + f"Connection error: {method} {path}", + retryable=True, + ) from e + except httpx.HTTPError as e: + raise IntercomError( + IntercomErrorCode.NETWORK_ERROR, + f"HTTP error: {method} {path}: {e}", + retryable=True, + ) from e + + if resp.status_code == 401: + raise IntercomError( + IntercomErrorCode.AUTH_ERROR, + f"Unauthorized: {resp.text}", + retryable=False, + ) + if resp.status_code == 403: + raise IntercomError( + IntercomErrorCode.AUTH_ERROR, + f"Forbidden: {resp.text}", + retryable=False, + ) + if resp.status_code == 404: + raise IntercomError( + IntercomErrorCode.NOT_FOUND, + f"Not found: {method} {path}", + retryable=False, + ) + if resp.status_code == 429: + retry_after = resp.headers.get("Retry-After") + raise IntercomError( + IntercomErrorCode.RATE_LIMITED, + f"Rate limited (Retry-After: {retry_after or 'N/A'}): {resp.text}", + retryable=True, + ) + if resp.status_code >= 500: + raise IntercomError( + IntercomErrorCode.NETWORK_ERROR, + f"Server error {resp.status_code}: {resp.text}", + retryable=True, + ) + if resp.status_code >= 400: + raise IntercomError( + IntercomErrorCode.UNKNOWN, + f"Client error {resp.status_code}: {resp.text}", + retryable=False, + ) + + remaining = resp.headers.get("X-RateLimit-Remaining") + limit = resp.headers.get("X-RateLimit-Limit") + if remaining is not None and int(remaining) < 10: + logger.warning("Intercom rate limit low: %s/%s", remaining, limit) + + try: + return resp.json() + except Exception: + return {} diff --git a/backend/package/yuxi/channel/extensions/intercom/config.py b/backend/package/yuxi/channel/extensions/intercom/config.py new file mode 100644 index 00000000..42656c7b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/intercom/config.py @@ -0,0 +1,192 @@ +import os +import logging + +from yuxi.channel.extensions.intercom.types import IntercomAccount + +logger = logging.getLogger(__name__) + + +class IntercomConfigAdapter: + def list_account_ids(self, config: dict) -> list[str]: + accounts = config.get("channels", {}).get("intercom", {}).get("accounts", {}) + if accounts: + return list(accounts.keys()) + if os.getenv("INTERCOM_ACCESS_TOKEN"): + return ["default"] + return [] + + async def resolve_account(self, account_id: str, config: dict | None = None) -> dict: + account = self._resolve_account_dataclass(account_id, config or {}) + return { + "account_id": account.account_id, + "access_token": account.access_token, + "webhook_secret": account.webhook_secret, + "admin_id": account.admin_id, + "dm_policy": account.dm_policy, + "allow_from": account.allow_from, + "auto_skip_closed": account.auto_skip_closed, + "auto_skip_snoozed": account.auto_skip_snoozed, + "handoff_policy": account.handoff_policy, + "datacenter": account.datacenter, + } + + def _resolve_account_dataclass(self, account_id: str, config: dict) -> IntercomAccount: + channel_cfg = config.get("channels", {}).get("intercom", {}) + accounts = channel_cfg.get("accounts", {}) + account_cfg = accounts.get(account_id, {}) if accounts else channel_cfg + + access_token = self._resolve_token(account_cfg.get("accessToken"), "INTERCOM_ACCESS_TOKEN") + webhook_secret = self._resolve_token(account_cfg.get("webhookSecret"), "INTERCOM_WEBHOOK_SECRET") + admin_id = account_cfg.get("adminId") or os.getenv("INTERCOM_ADMIN_ID", "") + + return IntercomAccount( + account_id=account_id, + access_token=access_token, + webhook_secret=webhook_secret, + admin_id=admin_id, + dm_policy=account_cfg.get("dmPolicy", "open"), + allow_from=account_cfg.get("allowFrom", []), + auto_skip_closed=account_cfg.get("autoSkipClosed", True), + auto_skip_snoozed=account_cfg.get("autoSkipSnoozed", True), + handoff_policy=account_cfg.get("handoffPolicy", "ignore"), + datacenter=account_cfg.get("datacenter", "us"), + ) + + @staticmethod + def _resolve_token(account_value: str | None, env_key: str) -> str: + if account_value is not None: + if account_value == "none": + return "" + if account_value.startswith("$"): + env_val = os.environ.get(account_value.strip("${}"), "") + return env_val + return account_value + return os.environ.get(env_key, "") + + def is_configured(self, account: dict) -> bool: + return bool(account.get("access_token")) + + def is_enabled(self, account: dict, config: dict) -> bool: + return True + + def disabled_reason(self, account: dict, config: dict) -> str: + return "" + + def unconfigured_reason(self, account: dict, config: dict) -> str: + if self.is_configured(account): + return "" + return "Missing required credential: accessToken" + + def describe_account(self, account: dict, config: dict) -> dict: + return { + "account_id": account.get("account_id", ""), + "dm_policy": account.get("dm_policy", "open"), + "handoff_policy": account.get("handoff_policy", "ignore"), + } + + def default_account_id(self, config: dict) -> str: + return config.get("channels", {}).get("intercom", {}).get("default_account", "default") + + def resolve_allow_from(self, config: dict, account_id: str | None = None) -> list[str] | None: + channel_cfg = config.get("channels", {}).get("intercom", {}) + accounts = channel_cfg.get("accounts", {}) + if account_id and account_id in accounts: + return accounts[account_id].get("allowFrom") + return channel_cfg.get("allowFrom") + + def format_allow_from(self, config: dict, account_id: str | None, allow_from: list[str | int]) -> list[str]: + return [str(entry) for entry in allow_from] + + def has_configured_state(self, config: dict) -> bool: + accounts = config.get("channels", {}).get("intercom", {}).get("accounts", {}) + if accounts: + for account_cfg in accounts.values(): + if account_cfg.get("accessToken") or account_cfg.get("accessToken") is None: + return True + return bool(os.getenv("INTERCOM_ACCESS_TOKEN")) + + def has_persisted_auth_state(self, config: dict) -> bool: + accounts = config.get("channels", {}).get("intercom", {}).get("accounts", {}) + if accounts: + for account_cfg in accounts.values(): + if account_cfg.get("accessToken"): + return True + return False + + def config_schema(self) -> dict: + return { + "$schema": "https://json-schema.org/draft-07/schema#", + "type": "object", + "title": "Intercom 渠道配置", + "properties": { + "accounts": { + "type": "object", + "title": "账户列表", + "description": "Intercom 账户配置，key 为账户 ID", + "additionalProperties": { + "type": "object", + "properties": { + "accessToken": { + "type": "string", + "title": "Access Token", + "description": "Intercom Access Token (Settings > Developer Hub > Access Tokens)", + "x-ui-password": True, + }, + "webhookSecret": { + "type": "string", + "title": "Webhook Secret", + "description": "用于验证 Webhook 请求的 HMAC 密钥", + "x-ui-password": True, + }, + "adminId": { + "type": "string", + "title": "Admin ID", + "description": "Bot 的管理员 ID（留空则自动从 API 获取）", + }, + "dmPolicy": { + "type": "string", + "title": "DM 安全策略", + "enum": ["open", "pairing", "allowlist", "disabled"], + "default": "open", + }, + "allowFrom": { + "type": "array", + "title": "白名单 Contact ID", + "items": {"type": "string"}, + "description": "仅 dmPolicy=allowlist 时生效，格式: contact_xxx 或 lead_xxx", + }, + "autoSkipClosed": { + "type": "boolean", + "title": "跳过已关闭会话", + "default": True, + }, + "autoSkipSnoozed": { + "type": "boolean", + "title": "跳过已暂停会话", + "default": True, + }, + "handoffPolicy": { + "type": "string", + "title": "人工接管策略", + "enum": ["ignore", "note_only", "continue"], + "default": "ignore", + "description": "检测到人工接管后的行为", + }, + "datacenter": { + "type": "string", + "title": "数据中心", + "enum": ["us", "eu", "au"], + "default": "us", + "description": "Intercom 数据中心区域", + }, + }, + "required": ["accessToken"], + }, + }, + "default_account": { + "type": "string", + "title": "默认账户", + "default": "default", + }, + }, + } diff --git a/backend/package/yuxi/channel/extensions/intercom/constants.py b/backend/package/yuxi/channel/extensions/intercom/constants.py new file mode 100644 index 00000000..af20ca5f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/intercom/constants.py @@ -0,0 +1,72 @@ +INTERCOM_API_BASES = { + "us": "https://api.intercom.io", + "eu": "https://api.eu.intercom.io", + "au": "https://api.au.intercom.io", +} +DEFAULT_DATACENTER = "us" +INTERCOM_API_VERSION = "2.15" + +INTERCOM_DEFAULT_TIMEOUT = 30.0 +INTERCOM_MAX_RETRIES = 3 +INTERCOM_RETRY_BASE_DELAY = 1.0 +INTERCOM_RETRY_MAX_DELAY = 8.0 + +INTERCOM_DEDUPE_TTL_SECONDS = 60 +INTERCOM_DEDUPE_MAX_SIZE = 10000 + +INTERCOM_TEXT_CHUNK_LIMIT = 40000 + +INTERCOM_BLOCK_STREAMING_CHUNK_MIN_CHARS = 500 +INTERCOM_BLOCK_STREAMING_CHUNK_MAX_CHARS = 2000 + +INTERCOM_WEBHOOK_TOPICS_HANDLED = frozenset( + [ + # 用户消息事件 + "conversation.user.created", + "conversation.user.replied", + # 管理员事件（P0 新增） + "conversation.admin.closed", + "conversation.admin.opened", + "conversation.admin.snoozed", + "conversation.admin.unsnoozed", + "conversation.admin.assigned", + "conversation.admin.replied", + "conversation.admin.single.created", + # 其他 Bot / Operator 回复（P0 新增） + "conversation.operator.replied", + # 会话状态事件（P1 新增） + "conversation.deleted", + "conversation.priority.updated", + "conversation.rating.added", + "conversation.read", + # 消息事件（P1 新增） + "conversation_part.redacted", + "conversation_part.tag.created", + ] +) + +INTERCOM_CONVERSATION_STATES = frozenset(["open", "snoozed", "closed"]) + +INTERCOM_PART_TYPES = frozenset(["comment", "note", "assignment", "close", "open", "snooze"]) + +INTERCOM_DM_POLICY_OPEN = "open" +INTERCOM_DM_POLICY_PAIRING = "pairing" +INTERCOM_DM_POLICY_ALLOWLIST = "allowlist" +INTERCOM_DM_POLICY_DISABLED = "disabled" + +INTERCOM_HANDOFF_POLICY_IGNORE = "ignore" +INTERCOM_HANDOFF_POLICY_NOTE_ONLY = "note_only" +INTERCOM_HANDOFF_POLICY_CONTINUE = "continue" + +INTERCOM_AGENT_PROMPT_RULES = [ + "Your response will be sent to Intercom Messenger as an admin reply.", + "The message format is HTML. Use for bold, for italic.", + 'Use text for links.', + "Use for unordered lists.", + "Use
  1. item
for ordered lists.", + "Use for inline code and 
 for code blocks.",
+    "Do not use markdown headings (use 
heading
 instead).",
+    "Do not use pipe tables (use 
 for table-like text).",
+    "Keep responses friendly, professional, and solution-oriented.",
+    "Identify yourself as an AI assistant when relevant.",
+]
diff --git a/backend/package/yuxi/channel/extensions/intercom/dedupe.py b/backend/package/yuxi/channel/extensions/intercom/dedupe.py
new file mode 100644
index 00000000..6b4f8476
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/dedupe.py
@@ -0,0 +1,62 @@
+import time
+import logging
+
+from yuxi.channel.extensions.intercom.constants import (
+    INTERCOM_DEDUPE_TTL_SECONDS,
+    INTERCOM_DEDUPE_MAX_SIZE,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class DedupeCache:
+    def __init__(self, ttl_seconds: int = INTERCOM_DEDUPE_TTL_SECONDS, max_size: int = INTERCOM_DEDUPE_MAX_SIZE):
+        self._ttl = ttl_seconds
+        self._max_size = max_size
+        self._cache: dict[str, float] = {}
+
+    def is_duplicate(self, event_id: str) -> bool:
+        if not event_id:
+            return False
+
+        now = time.monotonic()
+
+        if event_id in self._cache:
+            if now - self._cache[event_id] < self._ttl:
+                return True
+
+        self._cache[event_id] = now
+        self._evict_expired(now)
+        return False
+
+    def check_duplicate(self, event_id: str) -> bool:
+        if not event_id:
+            return False
+        now = time.monotonic()
+        if event_id in self._cache:
+            if now - self._cache[event_id] < self._ttl:
+                return True
+        return False
+
+    def mark_seen(self, event_id: str) -> None:
+        if not event_id:
+            return
+        self._cache[event_id] = time.monotonic()
+        self._evict_expired(time.monotonic())
+
+    def reset(self) -> None:
+        self._cache.clear()
+
+    def _evict_expired(self, now: float) -> None:
+        if len(self._cache) <= self._max_size:
+            expired = [k for k, v in self._cache.items() if now - v >= self._ttl]
+            for k in expired:
+                del self._cache[k]
+
+        if len(self._cache) > self._max_size:
+            sorted_keys = sorted(self._cache, key=self._cache.get)
+            for k in sorted_keys[: len(self._cache) - self._max_size]:
+                del self._cache[k]
+
+    def clear(self) -> None:
+        self._cache.clear()
diff --git a/backend/package/yuxi/channel/extensions/intercom/errors.py b/backend/package/yuxi/channel/extensions/intercom/errors.py
new file mode 100644
index 00000000..535c2713
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/errors.py
@@ -0,0 +1,51 @@
+from enum import StrEnum
+
+
+class IntercomErrorCode(StrEnum):
+    CONFIG_ERROR = "config_error"
+    AUTH_ERROR = "auth_error"
+    RATE_LIMITED = "rate_limited"
+    NOT_FOUND = "not_found"
+    NETWORK_ERROR = "network_error"
+    CONVERSATION_CLOSED = "conversation_closed"
+    CONVERSATION_SNOOZED = "conversation_snoozed"
+    HANDOFF_DETECTED = "handoff_detected"
+    SEND_FAILED = "send_failed"
+    UNKNOWN = "unknown"
+
+
+class IntercomError(Exception):
+    def __init__(self, code: IntercomErrorCode, message: str, retryable: bool = False):
+        self.code = code
+        self.retryable = retryable
+        super().__init__(f"[{code.value}] {message}")
+
+
+_ERROR_RETRY_MAP: dict[str, tuple[bool, float | None]] = {
+    "rate_limited": (True, None),
+    "service_unavailable": (True, 1.0),
+    "internal_server_error": (True, 3.0),
+    "timeout": (True, 0.5),
+    "network_error": (True, 1.0),
+    "not_found": (False, None),
+    "unauthorized": (False, None),
+    "forbidden": (False, None),
+    "bad_request": (False, None),
+}
+
+
+def classify_intercom_error(status_code: int, error_body: str = "") -> tuple[bool, float | None]:
+    if status_code == 429:
+        return (True, None)
+    if status_code >= 500:
+        return (True, 2.0)
+    if status_code in (401, 403):
+        return (False, None)
+    if status_code == 404:
+        return (False, None)
+    return (False, None)
+
+
+def is_retryable_status(status_code: int) -> bool:
+    retryable, _ = classify_intercom_error(status_code)
+    return retryable
diff --git a/backend/package/yuxi/channel/extensions/intercom/format.py b/backend/package/yuxi/channel/extensions/intercom/format.py
new file mode 100644
index 00000000..bb7950c1
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/format.py
@@ -0,0 +1,106 @@
+import html
+import re
+from html.parser import HTMLParser
+
+
+class HTMLToTextParser(HTMLParser):
+    def __init__(self):
+        super().__init__()
+        self._parts: list[str] = []
+        self._skip_depth = 0
+        self._current_href: str | None = None
+
+    def handle_starttag(self, tag, attrs):
+        if tag in ("script", "style", "head", "meta", "link"):
+            self._skip_depth += 1
+        if tag == "a":
+            for name, value in attrs:
+                if name == "href":
+                    self._current_href = value
+                    break
+
+    def handle_endtag(self, tag):
+        if tag in ("script", "style", "head", "meta", "link") and self._skip_depth > 0:
+            self._skip_depth -= 1
+        if tag in ("p", "br", "li", "div", "tr", "h1", "h2", "h3", "h4", "h5", "h6"):
+            self._parts.append("\n")
+        if tag == "a" and self._current_href:
+            self._parts.append(f" ({self._current_href})")
+            self._current_href = None
+
+    def handle_data(self, data):
+        if self._skip_depth == 0:
+            self._parts.append(data)
+
+    def get_text(self) -> str:
+        return "".join(self._parts).strip()
+
+
+def html_to_text(html_body: str) -> str:
+    parser = HTMLToTextParser()
+    parser.feed(html_body)
+    return parser.get_text()
+
+
+def markdown_to_html(md_text: str) -> str:
+    text = html.escape(md_text)
+
+    text = re.sub(r"```(\w+)?\n?(.*?)```", r"
\2
", text, flags=re.DOTALL)
+    text = re.sub(r"`([^`]+)`", r"\1", text)
+    text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
+    text = re.sub(r"\*(.+?)\*", r"\1", text)
+    text = re.sub(r"~~(.+?)~~", r"\1", text)
+    text = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r'\1', text)
+
+    lines = text.split("\n")
+    result: list[str] = []
+    list_buffer: list[str] = []
+    in_list = False
+
+    for line in lines:
+        stripped = line.strip()
+
+        ul_match = re.match(r"^[\-\*]\s+(.+)$", stripped)
+        ol_match = re.match(r"^\d+\.\s+(.+)$", stripped)
+
+        if ul_match:
+            if not in_list:
+                in_list = True
+                list_buffer = []
+            list_buffer.append(f"
  • {ul_match.group(1)}
    • ")
+            continue
+
+        if ol_match:
+            if not in_list:
+                in_list = True
+                list_buffer = []
+            list_buffer.append(f"
  • {ol_match.group(1)}
    • ")
+            continue
+
+        if in_list:
+            in_list = False
+            result.append(f"
      {''.join(list_buffer)}
    ")
+            list_buffer = []
+
+        heading_match = re.match(r"^#{1,6}\s+(.+)$", stripped)
+        if heading_match:
+            result.append(f"
    {heading_match.group(1)}
    ")
+            continue
+
+        if not stripped:
+            result.append("
    ")
+        elif stripped.startswith("<") and not stripped.startswith("{stripped}
    ")
+
+    if in_list and list_buffer:
+        result.append(f"
      {''.join(list_buffer)}
    ")
+
+    return "\n".join(result)
+
+
+def get_agent_prompt_rules() -> list[str]:
+    from yuxi.channel.extensions.intercom.constants import INTERCOM_AGENT_PROMPT_RULES
+
+    return INTERCOM_AGENT_PROMPT_RULES
diff --git a/backend/package/yuxi/channel/extensions/intercom/gateway.py b/backend/package/yuxi/channel/extensions/intercom/gateway.py
new file mode 100644
index 00000000..0d02c841
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/gateway.py
@@ -0,0 +1,128 @@
+import hashlib
+import hmac
+import logging
+
+from yuxi.channel.context import ChannelContext
+from yuxi.channel.extensions.intercom.client import IntercomClient
+from yuxi.channel.extensions.intercom.config import IntercomConfigAdapter
+from yuxi.channel.extensions.intercom.dedupe import DedupeCache
+from yuxi.channel.extensions.intercom.errors import IntercomError, IntercomErrorCode
+from yuxi.channel.extensions.intercom.types import IntercomAdminIdentity
+
+logger = logging.getLogger(__name__)
+
+
+class IntercomGateway:
+    delivery_mode = "direct"
+
+    def __init__(self):
+        self._accounts: dict[str, dict] = {}
+        self._dedupe = DedupeCache()
+
+    async def start(self, ctx: ChannelContext) -> object:
+        adapter = IntercomConfigAdapter()
+        config = getattr(ctx, "raw_config", None) or {}
+        account = await adapter.resolve_account(ctx.account_id, config)
+
+        if not account.get("access_token"):
+            raise IntercomError(IntercomErrorCode.CONFIG_ERROR, "access_token is required")
+
+        client = IntercomClient(account["access_token"], datacenter=account.get("datacenter", "us"))
+
+        identity = await self._auth_test(client, account)
+        ctx._intercom_identity = identity
+
+        self._accounts[ctx.account_id] = {
+            "account": account,
+            "client": client,
+            "identity": identity,
+        }
+
+        logger.info(
+            "Intercom gateway started for account %s (admin=%s, app=%s)",
+            ctx.account_id,
+            identity.admin_id,
+            identity.app_id,
+        )
+
+        return {
+            "account": account,
+            "identity": identity,
+            "client": client,
+        }
+
+    async def stop(self, ctx: ChannelContext) -> None:
+        entry = self._accounts.pop(ctx.account_id, None)
+        if entry is None:
+            return
+
+        await entry["client"].close()
+
+        for attr in ("_intercom_identity",):
+            try:
+                delattr(ctx, attr)
+            except AttributeError:
+                pass
+
+        logger.info("Intercom gateway stopped for account %s", ctx.account_id)
+
+    async def _auth_test(self, client: IntercomClient, account: dict) -> IntercomAdminIdentity:
+        try:
+            me = await client.get_me()
+        except Exception as e:
+            raise IntercomError(IntercomErrorCode.AUTH_ERROR, f"Token 验证失败: {e}") from e
+
+        admin_id = account.get("admin_id") or str(me.get("id", ""))
+        return IntercomAdminIdentity(
+            admin_id=admin_id,
+            app_id=str(me.get("app", {}).get("id_code", "")),
+            name=me.get("name", ""),
+            email=me.get("email", ""),
+        )
+
+    def verify_webhook_signature(self, body: bytes, x_hub_signature: str, webhook_secret: str) -> bool:
+        if not webhook_secret:
+            logger.warning("Webhook secret not configured, skipping signature verification")
+            return True
+
+        if not x_hub_signature:
+            return False
+
+        if not x_hub_signature.startswith("sha256="):
+            return False
+
+        received = x_hub_signature[7:]
+        computed = hmac.new(
+            webhook_secret.encode("utf-8"),
+            body,
+            hashlib.sha256,
+        ).hexdigest()
+
+        return hmac.compare_digest(computed, received)
+
+    def get_account(self, account_id: str) -> dict | None:
+        return self._accounts.get(account_id)
+
+    def get_client(self, account_id: str) -> IntercomClient | None:
+        entry = self._accounts.get(account_id)
+        return entry["client"] if entry else None
+
+    def get_identity(self, account_id: str) -> IntercomAdminIdentity | None:
+        entry = self._accounts.get(account_id)
+        return entry["identity"] if entry else None
+
+    def is_duplicate(self, event_id: str) -> bool:
+        return self._dedupe.is_duplicate(event_id)
+
+    def check_duplicate(self, event_id: str) -> bool:
+        return self._dedupe.check_duplicate(event_id)
+
+    def mark_seen(self, event_id: str) -> None:
+        self._dedupe.mark_seen(event_id)
+
+    def reset_dedupe(self) -> None:
+        self._dedupe.reset()
+
+    @staticmethod
+    async def resolve_gateway_auth_bypass_paths(config: dict) -> list[str]:
+        return ["/webhook/intercom"]
diff --git a/backend/package/yuxi/channel/extensions/intercom/handoff.py b/backend/package/yuxi/channel/extensions/intercom/handoff.py
new file mode 100644
index 00000000..f48ebd8a
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/handoff.py
@@ -0,0 +1,54 @@
+import logging
+
+from yuxi.channel.extensions.intercom.client import IntercomClient
+
+logger = logging.getLogger(__name__)
+
+
+class HandoffDetector:
+    @staticmethod
+    async def is_human_takeover(client: IntercomClient, conversation_id: str, bot_admin_id: str) -> bool:
+        try:
+            conv = await client.get_conversation(conversation_id)
+        except Exception:
+            logger.warning("Failed to fetch conversation %s for handoff check", conversation_id, exc_info=True)
+            return False
+
+        assignee = conv.get("assignee") or {}
+        if assignee and assignee.get("type") == "admin":
+            admin_id = str(assignee.get("id", ""))
+            if admin_id and admin_id != bot_admin_id:
+                return True
+
+        return False
+
+    @staticmethod
+    async def check_recent_admin_reply(client: IntercomClient, conversation_id: str, bot_admin_id: str) -> bool:
+        try:
+            parts = await client.get_conversation_parts(conversation_id)
+        except Exception:
+            logger.warning("Failed to fetch conversation parts for handoff check", exc_info=True)
+            return False
+
+        for part in reversed(parts):
+            author = part.get("author", {})
+            if author.get("type") == "admin" and str(author.get("id", "")) != bot_admin_id:
+                return True
+            if part.get("part_type") == "comment" and author.get("type") in ("contact", "user"):
+                break
+        return False
+
+    @staticmethod
+    async def get_last_admin_reply_time(client: IntercomClient | None, conversation_id: str, bot_admin_id: str) -> float | None:
+        if client is None:
+            return None
+        try:
+            parts = await client.get_conversation_parts(conversation_id)
+        except Exception:
+            return None
+        for part in reversed(parts):
+            author = part.get("author", {})
+            if author.get("type") == "admin" and str(author.get("id", "")) != bot_admin_id:
+                created_at = part.get("created_at", 0)
+                return float(created_at) if created_at else None
+        return None
diff --git a/backend/package/yuxi/channel/extensions/intercom/monitor.py b/backend/package/yuxi/channel/extensions/intercom/monitor.py
new file mode 100644
index 00000000..b062607c
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/monitor.py
@@ -0,0 +1,54 @@
+import logging
+
+from yuxi.channel.extensions.intercom.format import html_to_text
+from yuxi.channel.extensions.intercom.types import IntercomInboundEvent, IntercomAdminIdentity
+from yuxi.channel.message.models import GroupContext, MessageType, PeerInfo, UnifiedMessage
+from yuxi.channel.routing.models import PeerKind
+
+logger = logging.getLogger(__name__)
+
+
+class IntercomMonitor:
+
+    @staticmethod
+    def is_echo(event: IntercomInboundEvent, identity: IntercomAdminIdentity) -> bool:
+        return event.author_type in ("admin", "bot") and event.author_id == identity.admin_id
+
+    @staticmethod
+    def convert_event(
+        event: IntercomInboundEvent,
+        account_id: str,
+    ) -> UnifiedMessage | None:
+        body_text = html_to_text(event.body_html)
+        if not body_text and not event.attachments:
+            return None
+
+        message_type = _detect_type(event)
+
+        return UnifiedMessage(
+            msg_id=f"intercom:{event.message_id}",
+            channel_type="intercom",
+            account_id=account_id,
+            content=body_text,
+            sender=PeerInfo(
+                kind=PeerKind.DIRECT,
+                id=event.author_id,
+                display_name=event.author_name,
+            ),
+            message_type=message_type,
+            group=GroupContext(id=event.conversation_id),
+            raw_payload=event.raw,
+            timestamp=event.created_at if event.created_at else None,
+        )
+
+
+def _detect_type(event: IntercomInboundEvent) -> MessageType:
+    if event.attachments:
+        for att in event.attachments:
+            content_type = att.get("content_type", "")
+            if content_type.startswith("image/"):
+                return MessageType.IMAGE
+            if content_type.startswith("video/"):
+                return MessageType.VIDEO
+        return MessageType.FILE
+    return MessageType.TEXT
diff --git a/backend/package/yuxi/channel/extensions/intercom/outbound.py b/backend/package/yuxi/channel/extensions/intercom/outbound.py
new file mode 100644
index 00000000..5e823b6a
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/outbound.py
@@ -0,0 +1,371 @@
+import asyncio
+import logging
+import random
+
+from yuxi.channel.extensions.intercom.client import IntercomClient
+from yuxi.channel.extensions.intercom.config import IntercomConfigAdapter
+from yuxi.channel.extensions.intercom.constants import (
+    INTERCOM_MAX_RETRIES,
+    INTERCOM_RETRY_BASE_DELAY,
+    INTERCOM_RETRY_MAX_DELAY,
+    INTERCOM_TEXT_CHUNK_LIMIT,
+)
+from yuxi.channel.extensions.intercom.errors import IntercomError, IntercomErrorCode
+
+logger = logging.getLogger(__name__)
+
+
+class IntercomOutbound:
+    delivery_mode = "direct"
+    chunker_mode = "length"
+    text_chunk_limit = INTERCOM_TEXT_CHUNK_LIMIT
+    poll_max_options = None
+    supports_poll_duration_seconds = False
+    supports_anonymous_polls = False
+    extract_markdown_images = False
+    presentation_capabilities = None
+    delivery_capabilities = None
+
+    def __init__(self, gateway=None):
+        self._gateway = gateway
+
+    def chunker(self, text: str, limit: int, ctx: object | None = None) -> list[str]:
+        if not text:
+            return []
+        if len(text) <= limit:
+            return [text]
+
+        chunks: list[str] = []
+        paragraphs = text.split("\n\n")
+
+        for para in paragraphs:
+            if not para:
+                continue
+            if len(para) <= limit:
+                chunks.append(para)
+            else:
+                sentences = para.split("\n")
+                for sentence in sentences:
+                    if len(sentence) <= limit:
+                        chunks.append(sentence)
+                    else:
+                        while sentence:
+                            chunks.append(sentence[:limit])
+                            sentence = sentence[limit:]
+
+        merged: list[str] = []
+        for chunk in chunks:
+            if not merged:
+                merged.append(chunk)
+                continue
+            last = merged[-1]
+            candidate = last + "\n\n" + chunk
+            if len(candidate) <= limit:
+                merged[-1] = candidate
+            else:
+                merged.append(chunk)
+
+        return merged
+
+    async def _get_client(self, account_id: str | None) -> tuple[IntercomClient, dict]:
+        if self._gateway:
+            entry = self._gateway.get_account(account_id or "default")
+            if entry and entry.get("client"):
+                return entry["client"], entry["account"]
+
+        adapter = IntercomConfigAdapter()
+        config = {}
+        account = await adapter.resolve_account(account_id or "default", config)
+        client = IntercomClient(account["access_token"], datacenter=account.get("datacenter", "us"))
+        return client, account
+
+    def _owns_client(self, client: IntercomClient, account_id: str | None) -> bool:
+        if not self._gateway:
+            return True
+        entry = self._gateway.get_account(account_id or "default")
+        return not (entry and entry.get("client") is client)
+
+    async def _retry_with_backoff(self, operation, max_attempts: int | None = None):
+        max_attempts = max_attempts or INTERCOM_MAX_RETRIES
+
+        for attempt in range(max_attempts):
+            try:
+                return await operation()
+            except IntercomError as e:
+                if e.code == IntercomErrorCode.RATE_LIMITED and attempt < max_attempts - 1:
+                    delay = random.uniform(1, 4) * (2**attempt)
+                    logger.warning("Intercom rate limited, waiting %.1fs", delay)
+                    await asyncio.sleep(delay)
+                    continue
+
+                if e.code in (IntercomErrorCode.AUTH_ERROR, IntercomErrorCode.NOT_FOUND):
+                    raise
+
+                if attempt < max_attempts - 1:
+                    delay = min(
+                        INTERCOM_RETRY_BASE_DELAY * (2**attempt) + random.uniform(0, 1),
+                        INTERCOM_RETRY_MAX_DELAY,
+                    )
+                    logger.warning("Intercom API error, retrying in %.1fs", delay)
+                    await asyncio.sleep(delay)
+                    continue
+
+                raise
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        from yuxi.channel.extensions.intercom.format import markdown_to_html
+
+        html_body = markdown_to_html(content)
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+
+        try:
+            await self._retry_with_backoff(
+                lambda: client.reply_conversation(
+                    conversation_id=target_id,
+                    body=html_body,
+                    message_type="comment",
+                    reply_type="admin",
+                )
+            )
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def send_media(
+        self,
+        target_id: str,
+        media_url: str,
+        media_type: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+        attachment_urls = [media_url] if media_url else []
+
+        try:
+            await self._retry_with_backoff(
+                lambda: client.reply_conversation(
+                    conversation_id=target_id,
+                    body="",
+                    message_type="comment",
+                    reply_type="admin",
+                    attachment_urls=attachment_urls,
+                )
+            )
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def send_note(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        from yuxi.channel.extensions.intercom.format import markdown_to_html
+
+        html_body = markdown_to_html(content)
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+
+        try:
+            await self._retry_with_backoff(
+                lambda: client.reply_conversation(
+                    conversation_id=target_id,
+                    body=html_body,
+                    message_type="note",
+                    reply_type="admin",
+                )
+            )
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def send_quick_replies(
+        self,
+        target_id: str,
+        content: str,
+        options: list[dict],
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        from yuxi.channel.extensions.intercom.format import markdown_to_html
+
+        html_body = markdown_to_html(content)
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+
+        try:
+            reply_options = {
+                "type": "quick_replies",
+                "quick_reply_type": options[0].get("type", "button") if options else "button",
+                "quick_replies": [
+                    {"label": opt["label"], "value": opt.get("value", opt["label"])}
+                    for opt in options[:10]
+                ],
+            }
+            await self._retry_with_backoff(
+                lambda: client.reply_conversation(
+                    conversation_id=target_id,
+                    body=html_body,
+                    message_type="comment",
+                    reply_type="admin",
+                    reply_options=reply_options,
+                )
+            )
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def add_tags(
+        self,
+        target_id: str,
+        tags: list[str],
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+        try:
+            await self._retry_with_backoff(lambda: client.add_conversation_tags(target_id, tags))
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def remove_tags(
+        self,
+        target_id: str,
+        tag_ids: list[str],
+        *,
+        account_id: str | None = None,
+    ) -> None:
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+        try:
+            for tag_id in tag_ids:
+                await self._retry_with_backoff(lambda tid=tag_id: client.remove_conversation_tag(target_id, tid))
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def close_conversation(
+        self,
+        target_id: str,
+        *,
+        account_id: str | None = None,
+    ) -> bool:
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+        try:
+            await self._retry_with_backoff(
+                lambda: client.manage_conversation(target_id, "close", admin_id=account.get("admin_id"))
+            )
+            return True
+        except Exception:
+            logger.exception("Failed to close conversation %s", target_id)
+            return False
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def open_conversation(
+        self,
+        target_id: str,
+        *,
+        account_id: str | None = None,
+    ) -> bool:
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+        try:
+            await self._retry_with_backoff(
+                lambda: client.manage_conversation(target_id, "open", admin_id=account.get("admin_id"))
+            )
+            return True
+        except Exception:
+            logger.exception("Failed to open conversation %s", target_id)
+            return False
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def create_conversation(
+        self,
+        contact_id: str,
+        body: str,
+        *,
+        account_id: str | None = None,
+    ) -> dict | None:
+        from yuxi.channel.extensions.intercom.format import markdown_to_html
+
+        html_body = markdown_to_html(body)
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+        try:
+            result = await self._retry_with_backoff(
+                lambda: client.create_conversation(
+                    contact_id=contact_id,
+                    body=html_body,
+                    admin_id=account.get("admin_id"),
+                )
+            )
+            return result
+        except Exception:
+            logger.exception("Failed to create conversation for contact %s", contact_id)
+            return None
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def convert_to_ticket(
+        self,
+        conversation_id: str,
+        *,
+        account_id: str | None = None,
+    ) -> dict | None:
+        client, account = await self._get_client(account_id)
+        owns_client = self._owns_client(client, account_id)
+        try:
+            result = await self._retry_with_backoff(
+                lambda: client.convert_conversation_to_ticket(
+                    conversation_id,
+                    admin_id=account.get("admin_id"),
+                )
+            )
+            return result
+        except Exception:
+            logger.exception("Failed to convert conversation %s to ticket", conversation_id)
+            return None
+        finally:
+            if owns_client:
+                await client.close()
+
+    async def send_payload(self, ctx: object) -> object:
+        return None
+
+    async def send_poll(self, ctx: object) -> object:
+        return None
+
+    def sanitize_text(self, text: str, payload: object) -> str:
+        return text
+
+    def should_skip_plain_text_sanitization(self, payload: object) -> bool:
+        return False
+
+    def normalize_payload(self, payload, config, account_id=None):
+        return payload
+
+    def resolve_effective_text_chunk_limit(self, config, account_id=None, fallback_limit=None):
+        return fallback_limit or INTERCOM_TEXT_CHUNK_LIMIT
diff --git a/backend/package/yuxi/channel/extensions/intercom/pairing.py b/backend/package/yuxi/channel/extensions/intercom/pairing.py
new file mode 100644
index 00000000..dcaa6c3a
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/pairing.py
@@ -0,0 +1,43 @@
+import secrets
+import time
+import logging
+
+logger = logging.getLogger(__name__)
+
+_PAIRING_CODES: dict[str, tuple[str, float]] = {}
+_CODE_TTL_SECONDS = 300
+
+
+class IntercomPairing:
+    id_label = "intercomContactId"
+
+    async def generate_code(self, peer_id: str) -> str:
+        code = f"{secrets.randbelow(1_000_000):06d}"
+        _PAIRING_CODES[peer_id] = (code, time.monotonic())
+        logger.info("Intercom pairing code generated for contact %s", peer_id)
+        return code
+
+    async def verify_code(self, peer_id: str, code: str) -> bool:
+        stored = _PAIRING_CODES.get(peer_id)
+        if stored is None:
+            return False
+        stored_code, created_at = stored
+        if time.monotonic() - created_at > _CODE_TTL_SECONDS:
+            _PAIRING_CODES.pop(peer_id, None)
+            logger.info("Intercom pairing code expired for contact %s", peer_id)
+            return False
+        if not secrets.compare_digest(stored_code, code):
+            return False
+        _PAIRING_CODES.pop(peer_id, None)
+        logger.info("Intercom pairing code verified for contact %s", peer_id)
+        return True
+
+    def normalize_allow_entry(self, entry: str) -> str:
+        stripped = entry.strip()
+        for prefix in ("intercom:", "contact:"):
+            if stripped.lower().startswith(prefix):
+                stripped = stripped[len(prefix) :].strip()
+        return stripped
+
+    async def notify_approval(self, config: dict, peer_id: str, account_id: str | None = None) -> None:
+        logger.info("Intercom pairing approved for contact %s (account=%s)", peer_id, account_id or "default")
diff --git a/backend/package/yuxi/channel/extensions/intercom/plugin.json b/backend/package/yuxi/channel/extensions/intercom/plugin.json
new file mode 100644
index 00000000..777f8dfa
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/plugin.json
@@ -0,0 +1,28 @@
+{
+  "id": "intercom",
+  "name": "Intercom",
+  "version": "0.1.0",
+  "description": "Intercom Messenger channel plugin supporting direct chat, text/image/file messages, block streaming, and webhook-based inbound messaging",
+  "author": "ForcePilot",
+  "order": 65,
+  "dependencies": ["httpx"],
+  "capabilities": {
+    "chat_types": ["direct"],
+    "message_types": ["text", "image", "file"],
+    "interactions": ["button"],
+    "reactions": false,
+    "typing_indicator": false,
+    "threads": false,
+    "edit": false,
+    "unsend": false,
+    "reply": true,
+    "media": true,
+    "native_commands": false,
+    "polls": false,
+    "group_management": false,
+    "streaming": true,
+    "streaming_mode": "block",
+    "block_streaming": true
+  },
+  "enabled": true
+}
diff --git a/backend/package/yuxi/channel/extensions/intercom/security.py b/backend/package/yuxi/channel/extensions/intercom/security.py
new file mode 100644
index 00000000..f5f26a88
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/security.py
@@ -0,0 +1,66 @@
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class IntercomSecurity:
+    def resolve_dm_policy(self) -> dict:
+        return {"mode": "open", "allow_from": []}
+
+    async def check_allowlist(self, peer_id: str, channel_type: str) -> bool:
+        return True
+
+    def resolve_dm_policy_for_account(self, account: dict) -> dict:
+        mode = account.get("dm_policy", "open")
+        return {"mode": mode, "allow_from": account.get("allow_from", [])}
+
+    def authorize_dm(self, account: dict, contact_id: str) -> tuple[bool, str]:
+        policy = account.get("dm_policy", "open")
+        allow_from = account.get("allow_from", [])
+
+        if policy == "disabled":
+            return False, "DM is disabled"
+
+        if policy == "open":
+            if "*" in allow_from or not allow_from:
+                return True, ""
+            return contact_id in allow_from, "not-allowlisted"
+
+        if policy == "allowlist":
+            if not allow_from:
+                return False, "allowlist-empty"
+            return contact_id in allow_from, "not-allowlisted"
+
+        if policy == "pairing":
+            return True, "pairing-required"
+
+        return False, "unknown-policy"
+
+    def apply_config_fixes(self, config: dict) -> dict:
+        config.setdefault("channels", {})
+        config["channels"].setdefault("intercom", {})
+        return config
+
+    def collect_warnings(self, config: dict, account_id: str | None = None, account: dict | None = None) -> list[str]:
+        warnings: list[str] = []
+
+        if account:
+            if not account.get("webhook_secret"):
+                warnings.append("webhookSecret is not configured. Webhook signature verification will be skipped.")
+            if account.get("dm_policy") == "open" and not account.get("allow_from"):
+                warnings.append('dmPolicy is "open" without allowFrom. Consider adding allowFrom=["*"] explicitly.')
+            if account.get("dm_policy") == "allowlist" and not account.get("allow_from"):
+                warnings.append('dmPolicy is "allowlist" but allowFrom is empty. No contacts will be allowed.')
+
+        return warnings
+
+    def collect_audit_findings(
+        self,
+        config,
+        account_id=None,
+        account=None,
+        source_config=None,
+        ordered_account_ids=None,
+        has_explicit_account_path=False,
+    ) -> list[dict]:
+        return []
diff --git a/backend/package/yuxi/channel/extensions/intercom/session_guard.py b/backend/package/yuxi/channel/extensions/intercom/session_guard.py
new file mode 100644
index 00000000..8b6969f9
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/session_guard.py
@@ -0,0 +1,12 @@
+from yuxi.channel.extensions.intercom.types import IntercomInboundEvent
+
+
+class ConversationStateGuard:
+    @staticmethod
+    def should_respond(event: IntercomInboundEvent, account: dict) -> tuple[bool, str]:
+        state = event.conversation_state
+        if state == "closed" and account.get("auto_skip_closed", True):
+            return False, "conversation_closed"
+        if state == "snoozed" and account.get("auto_skip_snoozed", True):
+            return False, "conversation_snoozed"
+        return True, "ok"
diff --git a/backend/package/yuxi/channel/extensions/intercom/status.py b/backend/package/yuxi/channel/extensions/intercom/status.py
new file mode 100644
index 00000000..c3915015
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/status.py
@@ -0,0 +1,187 @@
+import time
+import logging
+
+from yuxi.channel.extensions.intercom.client import IntercomClient
+from yuxi.channel.extensions.intercom.types import IntercomProbeResult
+from yuxi.channel.protocols import ChannelAccountSnapshot, ChannelStatusIssue
+
+logger = logging.getLogger(__name__)
+
+
+class IntercomStatus:
+    default_runtime: ChannelAccountSnapshot | None = None
+
+    async def probe(self, account: dict | None = None) -> bool:
+        if not account:
+            return False
+        access_token = account.get("access_token", "")
+        if not access_token:
+            return False
+        probe_result = await self._probe_intercom(access_token)
+        return probe_result.ok
+
+    async def _probe_intercom(self, access_token: str, timeout: float = 10.0) -> IntercomProbeResult:
+        client = IntercomClient(access_token, timeout=timeout)
+        start = time.monotonic()
+
+        try:
+            me = await client.get_me()
+            latency_ms = (time.monotonic() - start) * 1000
+            return IntercomProbeResult(
+                ok=bool(me.get("id")),
+                admin_id=str(me.get("id", "")),
+                app_id=str(me.get("app", {}).get("id_code", "")),
+                name=me.get("name", ""),
+                email=me.get("email", ""),
+                latency_ms=latency_ms,
+            )
+        except Exception as e:
+            latency_ms = (time.monotonic() - start) * 1000
+            return IntercomProbeResult(ok=False, error=str(e), latency_ms=latency_ms)
+        finally:
+            await client.close()
+
+    def build_summary(self, snapshot: object) -> dict:
+        return {
+            "channel": "intercom",
+            "account_id": getattr(snapshot, "account_id", ""),
+            "status": getattr(snapshot, "status_state", "unknown"),
+        }
+
+    def build_channel_summary(
+        self,
+        account: dict,
+        config: dict,
+        default_account_id: str,
+        snapshot: ChannelAccountSnapshot,
+    ) -> dict:
+        account_id = account.get("account_id", "")
+        return {
+            "channel": "intercom",
+            "account_id": account_id,
+            "is_default": account_id == default_account_id,
+            "configured": snapshot.configured,
+            "enabled": snapshot.enabled,
+            "connected": snapshot.connected,
+            "running": snapshot.running,
+            "status": snapshot.status_state,
+            "health_state": snapshot.health_state,
+            "dm_policy": snapshot.dm_policy,
+            "last_error": snapshot.last_error,
+            "probe": snapshot.probe,
+        }
+
+    def build_account_snapshot(
+        self,
+        account: dict,
+        config: dict,
+        runtime: ChannelAccountSnapshot | None = None,
+        probe: object | None = None,
+        audit: object | None = None,
+    ) -> ChannelAccountSnapshot:
+        from yuxi.channel.extensions.intercom.config import IntercomConfigAdapter
+
+        adapter = IntercomConfigAdapter()
+        account_id = account.get("account_id", "")
+        configured = adapter.is_configured(account)
+        enabled = adapter.is_enabled(account, config)
+        status_state = self.resolve_account_state(account, config, configured, enabled)
+
+        return ChannelAccountSnapshot(
+            account_id=account_id,
+            enabled=enabled,
+            configured=configured,
+            status_state=status_state,
+            dm_policy=account.get("dm_policy", "open"),
+            allow_from=account.get("allow_from", []),
+            probe=probe,
+            audit=audit,
+        )
+
+    def format_capabilities_probe(self, probe: object) -> list[dict]:
+        if probe is None:
+            return []
+        return [
+            {
+                "label": "Intercom 连接",
+                "ok": getattr(probe, "ok", False),
+                "detail": f"{getattr(probe, 'name', '')} / {getattr(probe, 'email', '')}",
+                "latency_ms": getattr(probe, "latency_ms", None),
+            }
+        ]
+
+    async def audit_account(
+        self,
+        account: dict,
+        timeout_ms: int,
+        config: dict,
+        probe: object | None = None,
+    ) -> object:
+        access_token = account.get("access_token", "")
+        if not access_token:
+            return {"ok": False, "error": "No access token configured"}
+
+        probe_result = await self._probe_intercom(access_token)
+        return {
+            "ok": probe_result.ok,
+            "admin_id": probe_result.admin_id,
+            "app_id": probe_result.app_id,
+            "name": probe_result.name,
+            "email": probe_result.email,
+            "latency_ms": probe_result.latency_ms,
+            "error": probe_result.error,
+        }
+
+    async def build_capabilities_diagnostics(
+        self,
+        account: dict,
+        timeout_ms: int,
+        config: dict,
+        probe: object | None = None,
+        audit: object | None = None,
+        target: str | None = None,
+    ) -> dict | None:
+        return {
+            "channel": "intercom",
+            "probe": self.format_capabilities_probe(probe),
+            "audit": audit,
+        }
+
+    def log_self_id(
+        self,
+        account: dict,
+        config: dict,
+        runtime: object,
+        include_channel_prefix: bool = False,
+    ) -> None:
+        prefix = "Intercom " if include_channel_prefix else ""
+        account_id = account.get("account_id", "default")
+        logger.info("%sBot account loaded: id=%s", prefix, account_id)
+
+    def resolve_account_state(
+        self,
+        account: dict,
+        config: dict,
+        configured: bool,
+        enabled: bool,
+    ) -> str:
+        if not configured:
+            return "unconfigured"
+        if not enabled:
+            return "disabled"
+        return "enabled"
+
+    def collect_status_issues(self, accounts: list[ChannelAccountSnapshot]) -> list[ChannelStatusIssue]:
+        issues: list[ChannelStatusIssue] = []
+        for snapshot in accounts:
+            if snapshot.status_state == "unconfigured":
+                issues.append(
+                    ChannelStatusIssue(
+                        channel="intercom",
+                        account_id=snapshot.account_id,
+                        kind="unconfigured",
+                        message=f"Intercom 账户 {snapshot.account_id} 未配置凭证",
+                        fix="在管理面板中配置 Intercom Access Token",
+                    )
+                )
+        return issues
diff --git a/backend/package/yuxi/channel/extensions/intercom/streaming.py b/backend/package/yuxi/channel/extensions/intercom/streaming.py
new file mode 100644
index 00000000..268c9666
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/streaming.py
@@ -0,0 +1,67 @@
+import logging
+import time
+
+from yuxi.channel.extensions.intercom.constants import (
+    INTERCOM_BLOCK_STREAMING_CHUNK_MIN_CHARS,
+    INTERCOM_BLOCK_STREAMING_CHUNK_MAX_CHARS,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class IntercomStreaming:
+    streaming_mode = "block"
+    preview_stream_throttle_ms = 2000
+    preview_min_initial_chars = 100
+    block_streaming_enabled = True
+    block_streaming_break = "\n"
+    block_streaming_chunk_min_chars = INTERCOM_BLOCK_STREAMING_CHUNK_MIN_CHARS
+    block_streaming_chunk_max_chars = INTERCOM_BLOCK_STREAMING_CHUNK_MAX_CHARS
+    block_streaming_chunk_break_preference = "paragraph"
+    block_streaming_coalesce_defaults = None
+
+    def create_draft_stream_session(self, target_id: str) -> object:
+        return {"target_id": target_id, "content": ""}
+
+    def create_block_chunker(self) -> object:
+        return {"mode": "length"}
+
+    async def stream_reply(
+        self,
+        client,
+        conversation_id: str,
+        full_text_generator,
+        bot_admin_id: str,
+    ) -> str:
+        accumulated = ""
+        note_id = None
+        last_note_update = 0.0
+        NOTE_THROTTLE = 2.0
+
+        async for chunk in full_text_generator:
+            accumulated += chunk
+
+            now = time.monotonic()
+            if note_id is None:
+                try:
+                    resp = await client.reply_conversation(
+                        conversation_id,
+                        "正在生成回复...",
+                        message_type="note",
+                        reply_type="admin",
+                        admin_id=bot_admin_id,
+                    )
+                    note_id = resp.get("id", "")
+                except Exception:
+                    logger.warning("Failed to create draft note for streaming")
+
+            if now - last_note_update >= NOTE_THROTTLE and len(accumulated) > 100:
+                last_note_update = now
+
+        if note_id:
+            try:
+                await client.delete_conversation_part(conversation_id, str(note_id))
+            except Exception:
+                logger.warning("Failed to delete draft note %s", note_id)
+
+        return accumulated
diff --git a/backend/package/yuxi/channel/extensions/intercom/types.py b/backend/package/yuxi/channel/extensions/intercom/types.py
new file mode 100644
index 00000000..3e7ca2b5
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/types.py
@@ -0,0 +1,71 @@
+from dataclasses import dataclass, field
+from enum import StrEnum
+
+
+class IntercomDmPolicy(StrEnum):
+    OPEN = "open"
+    PAIRING = "pairing"
+    ALLOWLIST = "allowlist"
+    DISABLED = "disabled"
+
+
+class IntercomHandoffPolicy(StrEnum):
+    IGNORE = "ignore"
+    NOTE_ONLY = "note_only"
+    CONTINUE = "continue"
+
+
+class IntercomConversationState(StrEnum):
+    OPEN = "open"
+    SNOOZED = "snoozed"
+    CLOSED = "closed"
+
+
+@dataclass
+class IntercomAccount:
+    account_id: str
+    access_token: str = ""
+    webhook_secret: str = ""
+    admin_id: str = ""
+    dm_policy: str = "open"
+    allow_from: list[str] = field(default_factory=list)
+    auto_skip_closed: bool = True
+    auto_skip_snoozed: bool = True
+    handoff_policy: str = "ignore"
+    datacenter: str = "us"
+
+
+@dataclass
+class IntercomAdminIdentity:
+    admin_id: str = ""
+    app_id: str = ""
+    name: str = ""
+    email: str = ""
+
+
+@dataclass
+class IntercomInboundEvent:
+    event_id: str
+    topic: str
+    conversation_id: str
+    conversation_state: str = "open"
+    message_id: str = ""
+    author_type: str = ""
+    author_id: str = ""
+    author_name: str = ""
+    author_email: str = ""
+    body_html: str = ""
+    attachments: list[dict] = field(default_factory=list)
+    created_at: int = 0
+    raw: dict = field(default_factory=dict)
+
+
+@dataclass
+class IntercomProbeResult:
+    ok: bool
+    admin_id: str = ""
+    app_id: str = ""
+    name: str = ""
+    email: str = ""
+    latency_ms: float | None = None
+    error: str | None = None
diff --git a/backend/package/yuxi/channel/extensions/intercom/webhook.py b/backend/package/yuxi/channel/extensions/intercom/webhook.py
new file mode 100644
index 00000000..62c1ca53
--- /dev/null
+++ b/backend/package/yuxi/channel/extensions/intercom/webhook.py
@@ -0,0 +1,265 @@
+import asyncio
+import json
+import logging
+import time
+
+from fastapi import APIRouter, Request, Response
+from fastapi.responses import JSONResponse
+
+from yuxi.channel.extensions.intercom.constants import INTERCOM_WEBHOOK_TOPICS_HANDLED
+from yuxi.channel.extensions.intercom.format import html_to_text, markdown_to_html
+from yuxi.channel.extensions.intercom.gateway import IntercomGateway
+from yuxi.channel.extensions.intercom.handoff import HandoffDetector
+from yuxi.channel.extensions.intercom.monitor import IntercomMonitor
+from yuxi.channel.extensions.intercom.security import IntercomSecurity
+from yuxi.channel.extensions.intercom.session_guard import ConversationStateGuard
+from yuxi.channel.extensions.intercom.types import IntercomInboundEvent
+from yuxi.channel.runtime.manager import gateway
+from yuxi.channel.message.models import UnifiedMessage
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/webhook/intercom", tags=["intercom"])
+
+_gateway_instance: IntercomGateway | None = None
+_security = IntercomSecurity()
+_session_guard = ConversationStateGuard()
+_handoff = HandoffDetector()
+
+
+def set_gateway_instance(gw: IntercomGateway) -> None:
+    global _gateway_instance
+    _gateway_instance = gw
+
+
+def _parse_inbound_event(payload: dict) -> IntercomInboundEvent | None:
+    data = payload.get("data", {})
+    item = data.get("item", {})
+    topic = payload.get("topic", "")
+
+    # Admin / operator events have different payload structure
+    admin_topics = {
+        "conversation.admin.closed",
+        "conversation.admin.opened",
+        "conversation.admin.snoozed",
+        "conversation.admin.unsnoozed",
+        "conversation.admin.assigned",
+        "conversation.admin.replied",
+        "conversation.admin.single.created",
+        "conversation.operator.replied",
+    }
+    if topic in admin_topics:
+        return _parse_admin_event(payload)
+
+    # State-only events (deleted, priority updated, rating added, read)
+    state_only_topics = {
+        "conversation.deleted",
+        "conversation.priority.updated",
+        "conversation.rating.added",
+        "conversation.read",
+        "conversation_part.redacted",
+        "conversation_part.tag.created",
+    }
+    if topic in state_only_topics:
+        return _parse_state_event(payload)
+
+    conversation_message = item.get("conversation_message") or item.get("source") or {}
+    author = conversation_message.get("author", {})
+
+    if not conversation_message and item.get("conversation_parts"):
+        parts = item["conversation_parts"].get("conversation_parts", [])
+        if parts:
+            last_part = parts[-1]
+            conversation_message = last_part
+            author = last_part.get("author", {})
+
+    if not author.get("id"):
+        return None
+
+    return IntercomInboundEvent(
+        event_id=payload.get("id", ""),
+        topic=topic,
+        conversation_id=str(item.get("id", "")),
+        conversation_state=item.get("state", "open"),
+        message_id=str(conversation_message.get("id", "")),
+        author_type=author.get("type", ""),
+        author_id=str(author.get("id", "")),
+        author_name=author.get("name", ""),
+        author_email=author.get("email", ""),
+        body_html=conversation_message.get("body", ""),
+        attachments=conversation_message.get("attachments", []),
+        created_at=conversation_message.get("created_at", 0),
+        raw=payload,
+    )
+
+
+def _parse_admin_event(payload: dict) -> IntercomInboundEvent | None:
+    data = payload.get("data", {})
+    item = data.get("item", {})
+
+    topic = payload.get("topic", "")
+    admin = item.get("admin", {}) or item.get("assignee", {}) or item.get("operator", {})
+
+    return IntercomInboundEvent(
+        event_id=payload.get("id", ""),
+        topic=topic,
+        conversation_id=str(item.get("id", "")),
+        conversation_state=item.get("state", "open"),
+        message_id="",
+        author_type="admin",
+        author_id=str(admin.get("id", "")),
+        author_name=admin.get("name", ""),
+        author_email=admin.get("email", ""),
+        body_html="",
+        attachments=[],
+        created_at=0,
+        raw=payload,
+    )
+
+
+def _parse_state_event(payload: dict) -> IntercomInboundEvent | None:
+    data = payload.get("data", {})
+    item = data.get("item", {})
+
+    return IntercomInboundEvent(
+        event_id=payload.get("id", ""),
+        topic=payload.get("topic", ""),
+        conversation_id=str(item.get("id", "")),
+        conversation_state=item.get("state", "open"),
+        message_id="",
+        author_type="",
+        author_id="",
+        body_html="",
+        raw=payload,
+    )
+
+
+@router.post("/{account_id}")
+async def intercom_webhook(
+    request: Request,
+    account_id: str = "default",
+) -> Response:
+    if _gateway_instance is None:
+        return JSONResponse({"error": "Gateway not initialized"}, status_code=503)
+
+    raw_body = await request.body()
+
+    entry = _gateway_instance.get_account(account_id)
+    if not entry:
+        return JSONResponse({"error": "Account not found"}, status_code=404)
+
+    account = entry["account"]
+    webhook_secret = account.get("webhook_secret", "")
+    x_hub_signature = request.headers.get("X-Hub-Signature", "")
+
+    if not _gateway_instance.verify_webhook_signature(raw_body, x_hub_signature, webhook_secret):
+        logger.warning("Invalid webhook signature for account %s", account_id)
+        return JSONResponse({"error": "Invalid signature"}, status_code=401)
+
+    try:
+        payload = json.loads(raw_body)
+    except Exception:
+        return JSONResponse({"error": "Invalid JSON"}, status_code=400)
+
+    topic = payload.get("topic", "")
+    if topic not in INTERCOM_WEBHOOK_TOPICS_HANDLED:
+        return JSONResponse({"status": "ignored", "topic": topic})
+
+    # Handle state change events
+    state_change_topics = {
+        "conversation.admin.closed",
+        "conversation.admin.opened",
+        "conversation.admin.snoozed",
+        "conversation.admin.unsnoozed",
+    }
+    if topic in state_change_topics:
+        item = payload.get("data", {}).get("item", {})
+        logger.info("Conversation state change: %s for conv %s", topic, item.get("id"))
+        return JSONResponse({"status": "state_updated"})
+
+    event_id = payload.get("id", "")
+    if _gateway_instance.is_duplicate(event_id):
+        return JSONResponse({"status": "duplicate"})
+
+    inbound_event = _parse_inbound_event(payload)
+    if inbound_event is None:
+        return JSONResponse({"status": "ignored", "reason": "parse_failed"})
+
+    should_respond, reason = _session_guard.should_respond(inbound_event, account)
+    if not should_respond:
+        logger.info("Skipping event %s: %s (state=%s)", event_id, reason, inbound_event.conversation_state)
+        return JSONResponse({"status": "skipped", "reason": reason})
+
+    identity = _gateway_instance.get_identity(account_id)
+    if identity and IntercomMonitor.is_echo(inbound_event, identity):
+        return JSONResponse({"status": "ignored", "reason": "echo"})
+
+    allowed, auth_reason = _security.authorize_dm(account, inbound_event.author_id)
+    if not allowed:
+        logger.info("DM blocked for contact %s: %s", inbound_event.author_id, auth_reason)
+        return JSONResponse({"status": "blocked", "reason": auth_reason}, status_code=403)
+
+    handoff_policy = account.get("handoff_policy", "ignore")
+    client = _gateway_instance.get_client(account_id)
+    bot_admin_id = identity.admin_id if identity else ""
+    is_takeover = False
+
+    if client and bot_admin_id:
+        is_takeover = await _handoff.is_human_takeover(client, inbound_event.conversation_id, bot_admin_id)
+        if not is_takeover:
+            is_takeover = await _handoff.check_recent_admin_reply(client, inbound_event.conversation_id, bot_admin_id)
+
+    if is_takeover:
+        if handoff_policy == "ignore":
+            logger.info("Handoff detected for conv %s, policy=ignore, skipping", inbound_event.conversation_id)
+            return JSONResponse({"status": "skipped", "reason": "handoff_detected"})
+        if handoff_policy == "note_only":
+            logger.info("Handoff detected for conv %s, policy=note_only, sending note", inbound_event.conversation_id)
+            if client:
+                try:
+                    note_body = markdown_to_html(
+                        f"AI 建议回复（人工接管中）：\n\n{html_to_text(inbound_event.body_html)}"
+                    )
+                    await client.reply_conversation(
+                        inbound_event.conversation_id,
+                        note_body,
+                        message_type="note",
+                        reply_type="admin",
+                        admin_id=bot_admin_id,
+                    )
+                except Exception:
+                    logger.warning(
+                        "Failed to send handoff note for conv %s", inbound_event.conversation_id, exc_info=True
+                    )
+            return JSONResponse({"status": "note_only", "reason": "handoff_detected"})
+        if handoff_policy == "continue":
+            logger.info(
+                "Handoff detected for conv %s, policy=continue, still responding",
+                inbound_event.conversation_id,
+            )
+            last_ts = await _handoff.get_last_admin_reply_time(client, inbound_event.conversation_id, bot_admin_id)
+            if last_ts and time.time() - last_ts < 30:
+                logger.info("Recent human reply in conv %s, delaying AI response by 10s", inbound_event.conversation_id)
+                await asyncio.sleep(10)
+
+    processor = gateway._processor
+    if processor is not None:
+        msg = IntercomMonitor.convert_event(inbound_event, account_id)
+        if msg is not None:
+            asyncio.create_task(
+                _dispatch_to_agent(processor, msg),
+                name=f"intercom-agent-{account_id}-{inbound_event.conversation_id}",
+            )
+    else:
+        logger.warning("Message processor not available, cannot dispatch")
+
+    return JSONResponse({"status": "ok"})
+
+
+async def _dispatch_to_agent(processor, msg: UnifiedMessage) -> None:
+    try:
+        await asyncio.wait_for(processor.process(msg), timeout=120)
+    except TimeoutError:
+        logger.error("Agent response timeout for intercom:%s", msg.msg_id)
+    except Exception:
+        logger.exception("Failed to process Intercom message %s", msg.msg_id)