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 + an OPAQUE params dict forwarded verbatim to
  task/create; the platform does not interpret it), and 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) + the turn. Branches
  on the agent's ACP type: sync agents take the turn via message/send (reply returned
  inline); async/agentic via event/send (reply arrives on the task's stream).
- 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 (GitHub/Gitea). Generic JSON normalization.
- 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.
- Regenerate openapi.yaml for the new route.

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

Author: danielmillerp

agentex/openapi.yaml

@@ -364,6 +364,37 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /channels/webhook/{route_id}:
+    post:
+      tags:
+      - Channels
+      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.'
+      operationId: handle_webhook_channels_webhook__route_id__post
+      parameters:
+      - name: route_id
+        in: path
+        required: true
+        schema:
+          type: string
+          title: Route Id
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties: true
+                title: Response Handle Webhook Channels Webhook  Route Id  Post
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /tasks/{task_id}:
     get:
       tags:

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,109 @@
+"""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 and an opaque
+`params` dict forwarded verbatim to task/create (agentex does not interpret it —
+the bound agent does).
+
+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",
+        "params": {"system_prompt": "You are ...", "mcps": []}}}'
+"""
+
+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.domain.use_cases.agents_use_case import DAgentsUseCase
+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
+    # `params` is opaque — forwarded verbatim to task/create; agentex does not
+    # interpret it (an agent like golden-agent reads system_prompt/mcps/etc. there).
+    params = cfg.get("params")
+    return ChannelBinding(
+        secret=cfg["secret"],
+        agent_name=cfg["agent_name"],
+        params=params if isinstance(params, dict) else {},
+    )
+
+
+@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,
+    agents_use_case: DAgentsUseCase,
+) -> 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")
+
+    # Resolve the agent's ACP type so the router picks the right turn method
+    # (sync -> message/send returns the reply inline; async -> event/send).
+    agent = await agents_use_case.get(name=binding.agent_name)
+
+    inbound = _WEBHOOK.to_inbound(route_id, body)
+    result = await ChannelRouter(agents_acp_use_case).dispatch(
+        inbound, binding, agent.acp_type
+    )
+    response = {
+        "ok": True,
+        "channel": "webhook",
+        "route_id": route_id,
+        "task_id": result.task_id,
+    }
+    if result.reply is not None:
+        response["reply"] = result.reply
+    return response

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,88 @@
+"""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.
+
+    `params` is an OPAQUE dict forwarded verbatim as the task/create params — the
+    agentex platform does not interpret it. Whatever a given agent expects there
+    (e.g. golden-agent's system_prompt / mcps / harness / model) is that agent's
+    concern, not the channel layer's. Later this can be sourced from a saved config.
+    """
+
+    secret: str
+    agent_name: str
+    params: dict[str, Any] = field(default_factory=dict)
+    # Headers the router forwards to the agent (auth/delegation). Empty for local/open.
+    forward_headers: dict[str, str] = field(default_factory=dict)
+
+
+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,115 @@
+"""ChannelRouter: turn a normalized InboundMessage into an agent turn.
+
+Channel-agnostic — the same path serves webhook, Slack, or any future channel.
+Works for both ACP types: sync agents take the turn via message/send (reply returned
+inline); async/agentic agents via event/send (fire-and-forget, reply arrives on the
+task's message stream). Continuity is free: the InboundMessage's session_key is reused
+as the agentex task name (task/create is get-or-create on name).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from src.domain.channels.base import ChannelBinding, InboundMessage
+from src.domain.entities.agents import ACPType
+from src.domain.entities.agents_rpc import (
+    AgentRPCMethod,
+    CreateTaskRequestEntity,
+    SendEventRequestEntity,
+    SendMessageRequestEntity,
+)
+from src.domain.entities.task_messages import (
+    MessageAuthor,
+    TaskMessageContentType,
+    TextContentEntity,
+)
+from src.domain.use_cases.agents_acp_use_case import AgentsACPUseCase
+from src.utils.logging import make_logger
+
+logger = make_logger(__name__)
+
+
+@dataclass
+class DispatchResult:
+    task_id: str
+    # Sync agents return their reply inline; async agents reply later on the task
+    # stream, so reply is None and the caller can poll /messages if it wants it.
+    reply: str | None = None
+
+
+class ChannelRouter:
+    def __init__(self, acp_use_case: AgentsACPUseCase):
+        self._acp = acp_use_case
+
+    async def dispatch(
+        self, inbound: InboundMessage, binding: ChannelBinding, acp_type: ACPType
+    ) -> DispatchResult:
+        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.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 acp=%s -> task %s (session %s)",
+            inbound.channel,
+            inbound.route_id,
+            acp_type.value,
+            task_id,
+            session_key,
+        )
+
+        content = TextContentEntity(author=MessageAuthor.USER, content=inbound.text)
+
+        if acp_type == ACPType.SYNC:
+            # Sync: message/send carries the turn and returns the reply messages inline.
+            messages = await self._acp.handle_rpc_request(
+                method=AgentRPCMethod.MESSAGE_SEND,
+                params=SendMessageRequestEntity(
+                    task_id=task_id, content=content, stream=False
+                ),
+                agent_name=binding.agent_name,
+                request_headers=headers,
+            )
+            return DispatchResult(task_id=task_id, reply=_agent_text(messages))
+
+        # Async / agentic: event/send delivers the turn; the reply lands on the stream.
+        await self._acp.handle_rpc_request(
+            method=AgentRPCMethod.EVENT_SEND,
+            params=SendEventRequestEntity(task_id=task_id, content=content),
+            agent_name=binding.agent_name,
+            request_headers=headers,
+        )
+        return DispatchResult(task_id=task_id, reply=None)
+
+
+def _agent_text(messages: object) -> str | None:
+    """Join the agent-authored text from a sync message/send result."""
+    if not isinstance(messages, list):
+        return None
+    parts: list[str] = []
+    for m in messages:
+        content = getattr(m, "content", None)
+        if (
+            content is not None
+            and getattr(content, "type", None) == TaskMessageContentType.TEXT
+            and getattr(content, "author", None) == MessageAuthor.AGENT
+        ):
+            text = getattr(content, "content", "") or ""
+            if text.strip():
+                parts.append(text.strip())
+    return "\n\n".join(parts) if parts else None