feat(channels): add extensible channel abstraction with generic webhook ingress

Introduce a channel layer that normalizes external surfaces into agent turns,
so platforms beyond the chat UI can drive agents. The agent-driving core is
channel-agnostic; a new channel (e.g. Slack) implements one interface without
touching the router.

- domain/channels/base.py: InboundMessage, Channel ABC (authenticate + to_inbound),
  ChannelBinding (route -> agent + inline params), timing-safe shared-secret and
  HMAC-SHA256 verification helpers.
- domain/channels/router.py: ChannelRouter.dispatch() — InboundMessage ->
  task/create (get-or-create on a per-conversation session key) + event/send via
  the in-process ACP use case.
- domain/channels/webhook.py: WebhookChannel — generic HTTP ingress with a per-route
  secret. Accepts Authorization: Bearer / x-openclaw-webhook-secret, or an
  X-Hub-Signature-256 HMAC. Generic JSON normalization (no source-specific shaping).
- api/routes/channels.py: POST /channels/webhook/{route_id}. Route bindings load from
  the CHANNELS_WEBHOOK_ROUTES env var for now (seam for a future config store).
- Whitelist /channels/webhook in the auth middleware so it bypasses the agent
  API-key check and verifies its own per-route secret instead; register the router.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: danielmillerp

agentex/src/api/app.py

@@ -31,6 +31,7 @@
     agent_api_keys,
     agent_task_tracker,
     agents,
+    channels,
     checkpoints,
     deployment_history,
     deployments,
@@ -193,6 +194,7 @@ async def handle_unexpected(request, exc):
 
 # Include all routers
 fastapi_app.include_router(agents.router)
+fastapi_app.include_router(channels.router)
 fastapi_app.include_router(tasks.router)
 fastapi_app.include_router(messages.router)
 fastapi_app.include_router(spans.router)

agentex/src/api/middleware_utils.py

@@ -28,6 +28,9 @@
     "/readyz",
     "/ping",
     "/echo",
+    # Channels (webhook, …) bypass agentex API-key auth and verify a per-route
+    # shared secret / HMAC inside the channel instead.
+    "/channels/webhook",
 }
 
 DROP_HEADERS: set[str] = {

agentex/src/api/routes/channels.py

@@ -0,0 +1,92 @@
+"""Channels API: external surfaces (webhook now, Slack later) that drive agent turns.
+
+The webhook endpoint is auth-whitelisted in `middleware_utils.WHITELISTED_ROUTES`
+(it bypasses the agentex API-key auth) and instead verifies a per-route shared
+secret / HMAC inside the channel. Each route binds to one agent + inline config.
+
+Route bindings live in the CHANNELS_WEBHOOK_ROUTES env var (JSON) for now — the
+seam where a real per-config store (resolving a saved agent_config_id) plugs in later:
+
+    CHANNELS_WEBHOOK_ROUTES='{"demo": {"secret": "<shared-secret>",
+        "agent_name": "golden-agent", "system_prompt": "You are ...",
+        "harness": "claude-code", "model": "..."}}'
+"""
+
+from __future__ import annotations
+
+import json
+import os
+
+from fastapi import APIRouter, HTTPException, Request
+
+from src.domain.channels.base import ChannelBinding
+from src.domain.channels.router import ChannelRouter
+from src.domain.channels.webhook import MAX_BODY_BYTES, WebhookChannel
+from src.domain.use_cases.agents_acp_use_case import DAgentsACPUseCase
+from src.utils.logging import make_logger
+
+logger = make_logger(__name__)
+
+router = APIRouter(prefix="/channels", tags=["Channels"])
+
+# Channel registry — add "slack": SlackChannel() here when it lands.
+_WEBHOOK = WebhookChannel()
+
+
+def _webhook_binding(route_id: str) -> ChannelBinding | None:
+    """Resolve a webhook route's binding. Env-backed seam; replace with a config store."""
+    raw = os.environ.get("CHANNELS_WEBHOOK_ROUTES")
+    if not raw:
+        return None
+    try:
+        routes = json.loads(raw)
+    except json.JSONDecodeError:
+        logger.error("[channels] CHANNELS_WEBHOOK_ROUTES is not valid JSON")
+        return None
+    cfg = routes.get(route_id)
+    if not isinstance(cfg, dict) or not cfg.get("secret") or not cfg.get("agent_name"):
+        return None
+    return ChannelBinding(
+        secret=cfg["secret"],
+        agent_name=cfg["agent_name"],
+        system_prompt=cfg.get("system_prompt"),
+        harness=cfg.get("harness", "claude-code"),
+        model=cfg.get("model"),
+        allowed_tools=cfg.get("allowed_tools", []),
+        mcps=cfg.get("mcps", []),
+    )
+
+
+@router.post(
+    "/webhook/{route_id}",
+    summary="Generic webhook channel",
+    description="Authenticated webhook ingress: verifies a per-route secret/HMAC and "
+    "drives an agent turn. Auth is whitelisted at the middleware; the channel verifies "
+    "the route secret instead.",
+)
+async def handle_webhook(
+    route_id: str,
+    request: Request,
+    agents_acp_use_case: DAgentsACPUseCase,
+) -> dict:
+    binding = _webhook_binding(route_id)
+    if binding is None:
+        raise HTTPException(status_code=404, detail="unknown route")
+
+    raw = await request.body()
+    if len(raw) > MAX_BODY_BYTES:
+        raise HTTPException(status_code=413, detail="payload too large")
+    if not _WEBHOOK.authenticate(binding, request, raw):
+        raise HTTPException(status_code=401, detail="unauthorized")
+    if "application/json" not in request.headers.get("content-type", ""):
+        raise HTTPException(status_code=400, detail="expected application/json")
+    try:
+        body = json.loads(raw)
+    except json.JSONDecodeError:
+        raise HTTPException(status_code=400, detail="invalid json")
+    if not isinstance(body, dict):
+        raise HTTPException(status_code=400, detail="json body must be an object")
+
+    inbound = _WEBHOOK.to_inbound(route_id, body)
+    task_id = await ChannelRouter(agents_acp_use_case).dispatch(inbound, binding)
+    return {"ok": True, "channel": "webhook", "route_id": route_id, "task_id": task_id}

agentex/src/domain/channels/__init__.py

@@ -0,0 +1 @@
+"""Channels: normalize external surfaces (webhook, Slack, …) into agent turns."""

agentex/src/domain/channels/base.py

@@ -0,0 +1,95 @@
+"""Channel abstraction: normalize any external surface (webhook, Slack, …) into one
+inbound shape, and drive an agent turn from it.
+
+Modeled on OpenClaw's channel plugins (github.com/openclaw/openclaw,
+`src/channels/**`) and the claw0 tutorial: every platform produces the same
+`InboundMessage`; the agent-driving core is channel-agnostic. A new channel
+(Slack, etc.) implements `Channel` — `authenticate` + `to_inbound` for ingress,
+and (when it needs to push replies) its own outbound — without touching the router.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any
+
+from starlette.requests import Request
+
+
+@dataclass
+class InboundMessage:
+    """Normalized inbound event. Every channel produces this; the router only sees this."""
+
+    text: str
+    channel: str                       # "webhook" | "slack" | …
+    route_id: str = ""                 # which binding received it (webhook route / slack team)
+    peer_id: str = ""                  # conversation scope: DM user, channel:thread, route_id, …
+    sender_id: str = ""
+    raw: dict[str, Any] = field(default_factory=dict)
+
+    def session_key(self, agent_name: str) -> str:
+        """Stable per-conversation key → reused as the agentex task name (get-or-create)."""
+        basis = self.peer_id or self.route_id or "main"
+        digest = hashlib.sha1(f"{agent_name}:{self.channel}:{basis}".encode()).hexdigest()[:16]
+        return f"ch-{self.channel}-{digest}"
+
+
+@dataclass
+class ChannelBinding:
+    """A route's binding to one agent config. The 'configs later' seam — for now the
+    system prompt is inlined; later this resolves a saved agent_config_id."""
+
+    secret: str
+    agent_name: str
+    system_prompt: str | None = None
+    harness: str = "claude-code"
+    model: str | None = None
+    allowed_tools: list[str] = field(default_factory=list)
+    mcps: list[str] = field(default_factory=list)
+    # Headers the router forwards to the agent (auth/delegation). Empty for local/open.
+    forward_headers: dict[str, str] = field(default_factory=dict)
+
+    def task_params(self) -> dict[str, Any]:
+        params: dict[str, Any] = {"harness": self.harness}
+        if self.system_prompt:
+            params["system_prompt"] = self.system_prompt
+        if self.model:
+            params["model"] = self.model
+        if self.allowed_tools:
+            params["allowed_tools"] = self.allowed_tools
+        if self.mcps:
+            params["mcps"] = self.mcps
+        return params
+
+
+def verify_shared_secret(presented: str | None, secret: str) -> bool:
+    """Timing-safe shared-secret check (OpenClaw `safeEqualSecret`)."""
+    if not presented or not secret:
+        return False
+    return hmac.compare_digest(presented, secret)
+
+
+def verify_hmac_sha256(secret: str, body: bytes, signature_header: str | None) -> bool:
+    """Verify an `sha256=<hex>` HMAC signature (GitHub/Gitea-style)."""
+    if not signature_header or not signature_header.startswith("sha256="):
+        return False
+    expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+    return hmac.compare_digest(expected, signature_header)
+
+
+class Channel(ABC):
+    """One external surface. Ingress = authenticate + to_inbound. Outbound is
+    channel-specific (webhook replies in the HTTP response; Slack posts a message)."""
+
+    name: str = "unknown"
+
+    @abstractmethod
+    def authenticate(self, binding: ChannelBinding, request: Request, raw_body: bytes) -> bool:
+        """Return True iff the request is authentic for this binding."""
+
+    @abstractmethod
+    def to_inbound(self, route_id: str, body: dict[str, Any]) -> InboundMessage:
+        """Normalize the parsed JSON payload into an InboundMessage."""

agentex/src/domain/channels/router.py

@@ -0,0 +1,65 @@
+"""ChannelRouter: turn a normalized InboundMessage into an agent turn.
+
+Channel-agnostic — the same path serves webhook, Slack, or any future channel.
+Continuity is free: the InboundMessage's session_key is reused as the agentex task
+name (task/create is get-or-create on name), so repeated events on the same
+conversation continue the same task.
+"""
+
+from __future__ import annotations
+
+from src.domain.channels.base import ChannelBinding, InboundMessage
+from src.domain.entities.agents_rpc import (
+    AgentRPCMethod,
+    CreateTaskRequestEntity,
+    SendEventRequestEntity,
+)
+from src.domain.entities.task_messages import MessageAuthor, TextContentEntity
+from src.domain.use_cases.agents_acp_use_case import AgentsACPUseCase
+from src.utils.logging import make_logger
+
+logger = make_logger(__name__)
+
+
+class ChannelRouter:
+    def __init__(self, acp_use_case: AgentsACPUseCase):
+        self._acp = acp_use_case
+
+    async def dispatch(self, inbound: InboundMessage, binding: ChannelBinding) -> str:
+        """Get-or-create the conversation's task, then deliver the inbound text as an
+        event. Returns the task_id. Reply delivery is the channel's concern (sync HTTP
+        response, Slack post, …) and/or retrieved via the task's message stream."""
+        session_key = inbound.session_key(binding.agent_name)
+        headers = binding.forward_headers or None
+
+        task = await self._acp.handle_rpc_request(
+            method=AgentRPCMethod.TASK_CREATE,
+            params=CreateTaskRequestEntity(
+                name=session_key,
+                params=binding.task_params(),
+                task_metadata={
+                    "channel": inbound.channel,
+                    "route_id": inbound.route_id,
+                    "peer_id": inbound.peer_id,
+                    "sender_id": inbound.sender_id,
+                },
+            ),
+            agent_name=binding.agent_name,
+            request_headers=headers,
+        )
+        task_id = task.id
+        logger.info(
+            "[channels] %s route=%s -> task %s (session %s)",
+            inbound.channel, inbound.route_id, task_id, session_key,
+        )
+
+        await self._acp.handle_rpc_request(
+            method=AgentRPCMethod.EVENT_SEND,
+            params=SendEventRequestEntity(
+                task_id=task_id,
+                content=TextContentEntity(author=MessageAuthor.USER, content=inbound.text),
+            ),
+            agent_name=binding.agent_name,
+            request_headers=headers,
+        )
+        return task_id

agentex/src/domain/channels/webhook.py

@@ -0,0 +1,59 @@
+"""WebhookChannel: generic HTTP ingress (GitHub/Gitea/Zapier/anything that POSTs JSON).
+
+Modeled on OpenClaw `extensions/webhooks`: per-route shared secret, POST + JSON only,
+size cap, timing-safe auth. Accepts either:
+  - `Authorization: Bearer <secret>` / `x-openclaw-webhook-secret: <secret>`  (generic), or
+  - `X-Hub-Signature-256: sha256=<hmac>`                                       (GitHub/Gitea)
+
+The payload is normalized generically (text/message/goal/prompt, else the raw JSON).
+Source-specific shaping (e.g. PR rendering) belongs to the caller or the agent's
+system prompt — this stays a generic channel.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from starlette.requests import Request
+
+from src.domain.channels.base import (
+    Channel,
+    ChannelBinding,
+    InboundMessage,
+    verify_hmac_sha256,
+    verify_shared_secret,
+)
+
+MAX_BODY_BYTES = 256 * 1024
+
+
+class WebhookChannel(Channel):
+    name = "webhook"
+
+    def authenticate(self, binding: ChannelBinding, request: Request, raw_body: bytes) -> bool:
+        gh_sig = request.headers.get("x-hub-signature-256")
+        if gh_sig is not None:
+            return verify_hmac_sha256(binding.secret, raw_body, gh_sig)
+        auth = request.headers.get("authorization", "")
+        presented = auth[len("Bearer "):].strip() if auth.startswith("Bearer ") else None
+        presented = presented or request.headers.get("x-openclaw-webhook-secret")
+        return verify_shared_secret(presented, binding.secret)
+
+    def to_inbound(self, route_id: str, body: dict[str, Any]) -> InboundMessage:
+        return InboundMessage(
+            text=_render_text(body),
+            channel=self.name,
+            route_id=route_id,
+            peer_id=route_id,
+            sender_id="webhook",
+            raw=body,
+        )
+
+
+def _render_text(body: dict[str, Any]) -> str:
+    for key in ("text", "message", "goal", "prompt"):
+        value = body.get(key)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+    return json.dumps(body, indent=2)[:8000]