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 WhatsApp adapter (ak-py/src/agentkernel/integration/whatsapp/adapter.py) as the canonical webhook reference, and Gmail (integration/gmail/adapter.py) as the polling one.

Architecture Overview

A platform integration is two pure translation functions with a queue between them (spec #524):

1. An InboundAdapter turns one platform delivery into normalized InboundRequest envelopes: it verifies the delivery, extracts the text, downloads and stores attachments, and resolves session_id and request_id at the edge. It never runs the agent.
2. The pipeline carries the request: IntegrationProducer enqueues it, AgentRunner executes it platform-agnostically, and the reply travels back on the output queue with the integration attribute and the reply_-prefixed reply context.
3. An OutboundAdapter turns the agent's reply back into platform API calls, using nothing but the flat reply_context the inbound half resolved.

This is why the webhook answers in milliseconds: a slow agent run can no longer hold the turn open past the platform's delivery timeout and cause a redelivery.

Hosting depends on how the platform delivers events:

Adapters must be mounted inside the pipeline. WebhookRESTRequestHandler sets requires_pipeline = True, so RESTAPI.run([...]) refuses it with an AKConfigError: without a queue there would be no runner to drain what it enqueues, and the platform would get its 200 while the user never got a reply.

Step-by-Step

1. Create the Integration Directory

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

2. Implement the Inbound Adapter

# ak-py/src/agentkernel/integration/<platform>/adapter.py
import logging
from typing import Any, Dict, List, Optional

from fastapi import HTTPException, Request

from ...core.config import AKConfig
from ...core.model import AgentReply, AgentRequest, AgentRequestImage, AgentRequestText
from ...core.multimodal.storage import AttachmentStorageManager
from ..adapter.base import (
    ATTACHMENTS_DISABLED_ERROR,
    SESSION_CACHE_ERROR,
    InboundAdapter,
    InboundParseResult,
    InboundRequest,
    OutboundAdapter,
)
from ..adapter.routes import BUILTIN_WEBHOOK_ROUTES

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


class <Platform>InboundAdapter(InboundAdapter):
    """<Platform> deliveries -> normalized requests."""

    name = NAME
    # A built-in reads its paths from the route table (see step 7), so the Lambda authorizer's
    # WebhookRouteMatcher lets exactly this route through; a bring-your-own adapter spells it out.
    webhook_path = BUILTIN_WEBHOOK_ROUTES[NAME].webhook_path
    challenge_path = BUILTIN_WEBHOOK_ROUTES[NAME].challenge_path  # None unless the platform has a GET handshake

    _log = _log

    def __init__(self):
        config = AKConfig.get()
        self._agent = config.<platform>.agent or None
        self._max_file_size = config.api.max_file_size
        self._client = <Platform>Client()   # your API wrapper

    async def verify(self, raw: Request) -> None:
        """Reject a delivery that did not come from the platform. Runs before parse."""
        if not self._client.verify(await raw.body(), raw.headers.get("x-platform-signature", "")):
            raise HTTPException(status_code=403, detail="Invalid signature")

    async def parse(self, raw: Request) -> InboundParseResult:
        """One delivery can carry several messages; return one InboundRequest per message."""
        body = await raw.json()
        requests = [r for r in [await self._to_request(m) for m in body.get("messages", [])] if r is not None]
        return InboundParseResult(requests=requests)

    async def _to_request(self, message: dict) -> Optional[InboundRequest]:
        text = message.get("text", "")
        sender = message["from"]
        if not text:
            return None   # legitimately ignored: an empty list is not an error

        requests: List[AgentRequest] = [AgentRequestText(prompt=text)]
        # ... download attachments into `requests` here (see step 5) ...

        requests, _ = AttachmentStorageManager.offload(
            sender,
            requests,
            attachments_disabled_error=ATTACHMENTS_DISABLED_ERROR,
            session_cache_error=SESSION_CACHE_ERROR,
        )
        return InboundRequest(
            session_id=sender,             # the platform's conversation key
            request_id=message["id"],      # the platform's own id: this is what dedupes a retry
            requests=requests,
            prompt=text,
            agent=self._agent,
            user_id=sender,
            reply_context={"to": sender},  # flat, string-valued delivery coordinates
        )

Rules the adapter must hold to:

1. Never execute. No ChatService, AgentService or Runtime import. The only side effects allowed are platform API calls and attachment storage.
2. Read only your own config block, AKConfig.get().<platform>.
3. verify before parse, parse before enqueue. verify is concrete and a no-op on the base: override it only when verification is separable from parsing. (Slack and Teams verify inside their SDK's dispatch, so theirs stays the default.)
4. request_id is the platform's message id wherever one exists; that is what makes a webhook retry deduplicate instead of running the agent twice. Synthesize a stable one only when the platform gives you none (Slack: f"slack:{channel}:{ts}").
5. An ignored delivery returns an empty request list, never an exception.
6. Override missing_verification_settings() when verify skips its check for an unset secret (return e.g. ["<platform>.webhook_secret"] while it is empty). LambdaWebhookHost refuses such an adapter on cold start, because behind the authorizer's integration bypass the adapter's own check is the only one. An adapter that refuses to construct without its secret (Slack, Teams) needs no override.

3. Implement the Outbound Adapter

class <Platform>OutboundAdapter(OutboundAdapter):
    """Agent replies -> <Platform> messages."""

    name = NAME
    MESSAGE_LIMIT = 4096          # the platform's per-message limit; split_reply chunks to it
    MAX_CHUNKS = None             # or a cap, with TRUNCATION_NOTICE appended past it

    _log = _log

    def __init__(self):
        self._client = <Platform>Client()

    async def acknowledge(self, reply_context: Dict[str, str]) -> Dict[str, str]:
        """Edge-side feedback: a typing indicator, a read receipt, a "thinking" message.

        The returned dict is merged into reply_context, which is how Slack carries the id of
        its placeholder message through to delivery.
        """
        await self._client.typing(reply_context["to"])
        return {}

    async def deliver(self, reply: AgentReply, reply_context: Dict[str, str]) -> None:
        """Raising hands the message back for retry, then deliver_error."""
        await self._client.send(reply_context["to"], self.split_reply(str(reply)))

    async def deliver_error(self, message: str, reply_context: Dict[str, str]) -> None:
        try:
            await self._client.send(reply_context["to"], [message])
        except Exception as e:
            self._log.error(f"Could not deliver the <Platform> error message: {e}")

• The reply always arrives as an AgentReplyText: the Agent Runner serializes the typed reply to its string form before the output queue.
• Outbound adapters are cached and shared across consumer threads, and each call runs on its own event loop. Keep no per-message state on self, and build loop-bound clients (an httpx.AsyncClient) per call.
• deliver_error receives OutboundAdapter.ERROR_MESSAGE, not the raw exception: raw error text is logged, never sent to a platform user.

4. Reply Context

reply_context is flat, string-valued delivery coordinates: everything deliver needs and nothing else. It travels as reply_-prefixed message attributes rather than body fields, because BaseRunRequest is extra="allow" and an unknown body field would reach the agent as AgentRequestAny context.

Budget: 8 KB serialized, enforced in IntegrationProducer with a ValueError naming the adapter. If the platform's reply address is an object rather than strings, JSON-encode it into one value (Teams does this with its ConversationReference).

5. Attachments

Attachment bytes must not ride the queue: brokers cap a message far below api.max_file_size. Download at the edge (that is where the platform token is), then call AttachmentStorageManager.offload, which stores the bytes in the AttachmentStore and replaces each image/file request with an AgentRequestAttachmentRef.

This makes multimodal.enabled: true a requirement for attachment-bearing messages, and rejects multimodal.storage_type: session_cache (it writes into a session copy the runner process never sees). Both messages are shared constants; pass them through as shown in step 2.

6. Create the __init__.py and the Public Alias

# ak-py/src/agentkernel/integration/<platform>/__init__.py
from .adapter import <Platform>InboundAdapter, <Platform>OutboundAdapter

Create ak-py/src/agentkernel/<platform>.py with a wildcard import (see ak-py/src/agentkernel/slack.py):

from .integration.<platform> import *

7. Register the Built-in with the Factory

The Response Handler holds only the integration attribute string, so the outbound half is resolved by name. Add the platform to IntegrationAdapterFactory (integration/adapter/factory.py): its short name in _BUILTIN_NAMES, and an if/elif branch in _builtin importing the class inside require_extra.

(The inbound half is never resolved by name: the application constructs it and hands it to a host, so bring-your-own inbound is just passing a different instance.)

Add the platform's routes to BUILTIN_WEBHOOK_ROUTES in integration/adapter/routes.py as well, a WebhookRoute(name, webhook_path, challenge_path=...). It is the one definition the adapter's path attributes and WebhookRouteMatcher.for_integrations(...) both read, and IntegrationAdapterContract.test_a_builtin_is_served_where_the_authorizer_expects fails when they drift.