ak-dev-new-messaging-integration

Adding a New Messaging Integration

This guide walks through adding a new messaging platform integration to Agent Kernel. Use the Slack integration (ak-py/src/agentkernel/integration/slack/) as the canonical reference.

Architecture Overview

Messaging integrations follow a consistent pattern:

1. A request handler class that extends RESTRequestHandler
2. The handler exposes FastAPI routes for webhooks
3. Incoming messages are parsed into AgentRequest models (platform attachments downloaded and base64-encoded by the handler)
4. The ChatService execution core runs the agent: build a BaseChatRequest and call execute(req, requests=<prebuilt list>), which returns the typed reply. Integrations own their transport and reply formatting, so they call the core, never the HTTP-shaped process_* wrappers and never AgentService directly (see the chat execution layering rubric in ak-dev-architecture)
5. The reply is formatted and sent back via the platform's API; a ValueError from execute maps to the platform's "no agent available" message
6. Configuration is added to AKConfig (accessed via the Config.get() alias) for platform-specific settings

Exception: Gmail does not follow the webhook pattern. AgentGmailRequestHandler (integration/gmail/gmail_chat.py) has no base class and polls email via OAuth instead of exposing webhook routes; its config (_GmailConfig in core/config.py) has token_file, poll_interval, and label_filter rather than a webhook secret.

Step-by-Step

1. Create the Integration Directory

ak-py/src/agentkernel/integration/<platform>/
├── __init__.py
└── <platform>_chat.py

2. Implement the Request Handler

# ak-py/src/agentkernel/integration/<platform>/<platform>_chat.py
import logging
from agentkernel.api.handler import RESTRequestHandler
from agentkernel.core import ChatService, Config
from agentkernel.core.model import AgentRequestText, AgentRequestImage, AgentRequestFile, BaseChatRequest
from fastapi import APIRouter, Request

logger = logging.getLogger("ak.integration.<platform>")


class Agent<Platform>RequestHandler(RESTRequestHandler):
    """Handles incoming messages from <Platform> and routes them to Agent Kernel agents."""

    def __init__(self):
        config = Config.get().<platform>
        self._agent_name = config.agent if config else None
        self._chat_service = ChatService()
        # Initialize platform-specific client/SDK here
        # e.g., self._client = PlatformClient(token=config.bot_token)

    def get_router(self) -> APIRouter:
        router = APIRouter()

        @router.get("/health")
        async def health():
            return {"status": "ok"}

        @router.post("/<platform>/webhook")
        async def webhook(request: Request):
            body = await request.json()
            await self._handle_message(body)
            return {"status": "ok"}

        return router

    async def _handle_message(self, body: dict):
        """Parse platform message and route to agent."""
        # 1. Extract message content from platform-specific format
        user_id = body.get("user_id", "unknown")
        text = body.get("text", "")
        attachments = body.get("attachments", [])

        # 2. Build request list
        requests = []
        if text:
            requests.append(AgentRequestText(prompt=text))

        for attachment in attachments:
            # Handle images
            if attachment.get("type") == "image":
                image_data = await self._download_file(attachment["url"])
                requests.append(AgentRequestImage(
                    image_data=image_data,
                    name=attachment.get("name", "image"),
                    mime_type=attachment.get("mime_type")
                ))
            # Handle files
            elif attachment.get("type") == "file":
                file_data = await self._download_file(attachment["url"])
                requests.append(AgentRequestFile(
                    file_data=file_data,
                    name=attachment.get("name", "file"),
                    mime_type=attachment.get("mime_type")
                ))

        if not requests:
            return

        # 3. Run through the ChatService execution core with the prebuilt request list
        #    (prompt may be empty for attachment-only messages; user_id/group_id are
        #    best-effort platform identity)
        req = BaseChatRequest(prompt=text, agent=self._agent_name, session_id=user_id, user_id=user_id)
        try:
            reply, _ = await self._chat_service.execute(req, requests=requests)
        except ValueError as ve:
            logger.warning(f"Agent execution rejected: {ve}")
            await self._send_reply(user_id, "Sorry, no agent is available to handle your request.")
            return

        # 4. Send reply back via platform API
        await self._send_reply(user_id, str(reply))

    async def _download_file(self, url: str) -> str:
        """Download a file and return base64-encoded content."""
        import base64
        import httpx
        async with httpx.AsyncClient() as client:
            response = await client.get(url, headers=self._auth_headers())
            return base64.b64encode(response.content).decode("utf-8")

    async def _send_reply(self, channel: str, text: str):
        """Send reply back to the messaging platform."""
        # Platform-specific API call to send message
        # e.g., self._client.send_message(channel=channel, text=text)
        pass

    def _auth_headers(self) -> dict:
        """Return auth headers for platform API calls."""
        return {}