feat(sdk): add webhook helper for forward-route handlers

Add agentex.lib.sdk.utils.webhooks.handle_webhook — a reusable helper an agent
calls from a forward-route handler (@acp.post on the route the server's
/agents/forward/name/{agent}/{path} ingress proxies to). It shapes the payload
(generic or GitHub PR), resolves task params (inline or fetched from a config
resolve URL for config-by-id), get-or-creates a task on a stable session key so
repeat events fold into one task, drives the turn (sync message / async event),
and returns/polls the reply.

This keeps webhook triggering on the supported forward mechanism + its built-in
GitHub/Slack signature auth, instead of a parallel ingress. Config-by-id is
ingress-independent: point params_source at the platform's config-resolve endpoint.

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

Author: danielmillerp

src/agentex/lib/sdk/utils/webhooks.py

@@ -0,0 +1,311 @@
+"""Drive an agent turn from an inbound webhook, inside a forward-route handler.
+
+The Agentex server already exposes a webhook ingress: a request to
+``/agents/forward/name/{agent}/{path}`` is signature-verified (GitHub ``sha256=`` /
+Slack ``v0:`` HMAC via the agent's registered keys) and proxied to the agent's own
+HTTP route. This helper is what that route handler calls to turn the inbound payload
+into an agent turn — without each agent re-implementing payload shaping, config
+resolution, session continuity, and reply handling.
+
+Typical use inside an agent::
+
+    from fastapi import Request
+    from agentex.lib.sdk.utils.webhooks import handle_webhook
+
+
+    @acp.post("/github-pr")
+    async def github_pr(request: Request):
+        body = await request.json()
+        result = await handle_webhook(
+            agent_name="my-agent",
+            payload=body,
+            acp_type="sync",
+            shaper="github_pr",
+            params_source="https://<host>/public/v5/agent_configs/<id>/resolve",
+            params_source_headers={"x-api-key": ..., "x-selected-account-id": ...},
+            wait=True,
+        )
+        return {"task_id": result.task_id, "reply": result.reply}
+
+Config-by-id: pass ``params_source`` pointing at the platform's config-resolve
+endpoint; the resolved params (e.g. system_prompt / harness / model / tools) are
+forwarded opaquely to ``task/create``. Or pass inline ``params`` for a one-off.
+"""
+
+from __future__ import annotations
+
+import json
+import hashlib
+from typing import Any, Literal
+from dataclasses import field, dataclass
+from collections.abc import Callable, Awaitable
+
+from agentex.lib import adk
+from agentex.lib.utils.logging import make_logger
+from agentex.types.task_message_content import TextContent
+
+logger = make_logger(__name__)
+
+# Injectable params fetcher (url -> JSON). Default uses httpx; tests inject a fake.
+ParamsFetcher = Callable[[str], Awaitable[dict[str, Any]]]
+
+MAX_BODY_CHARS = 4000
+MAX_DIFF_CHARS = 30000
+
+
+class WebhookError(RuntimeError):
+    """Raised when a webhook turn cannot be driven (e.g. params resolution failed)."""
+
+
+@dataclass
+class WebhookResult:
+    task_id: str
+    # Sync agents reply inline. For async agents, ``reply`` is None unless ``wait`` was
+    # set, in which case it is the polled reply (or None if it didn't settle in time).
+    reply: str | None = None
+    task_metadata: dict[str, str] = field(default_factory=dict)
+
+
+# --------------------------------------------------------------------------- shaping
+
+
+def session_key(agent_name: str, channel: str, peer_id: str) -> str:
+    """Stable per-conversation task name → reused for get-or-create on task/create, so
+    repeat events from the same source fold into one task instead of spawning new ones."""
+    basis = peer_id or "main"
+    digest = hashlib.sha1(f"{agent_name}:{channel}:{basis}".encode()).hexdigest()[:16]
+    return f"wh-{channel}-{digest}"
+
+
+def render_generic(body: dict[str, Any]) -> str:
+    """Generic payload → prompt text: first of text/message/goal/prompt, else raw JSON."""
+    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]
+
+
+def shape_github_pr(body: dict[str, Any]) -> tuple[str, str | None, str]:
+    """Shape a GitHub/Gitea pull-request webhook into (prompt, peer_id, sender).
+
+    ``peer_id`` is ``repo#number`` so repeated events for the same PR (opened,
+    synchronize, ...) fold into one task. Falls back to generic rendering for non-PR
+    payloads (ping, issue, ...).
+    """
+    pull_request = body.get("pull_request")
+    if not isinstance(pull_request, dict):
+        return render_generic(body), None, _github_actor(body)
+
+    repo = _repo_full_name(body)
+    number = pull_request.get("number")
+    title = (pull_request.get("title") or "").strip()
+    action = (body.get("action") or "").strip()
+    description = (pull_request.get("body") or "").strip()
+    html_url = pull_request.get("html_url") or pull_request.get("url")
+
+    header = "Pull request"
+    if repo and number is not None:
+        header = f"Pull request {repo}#{number}"
+    elif number is not None:
+        header = f"Pull request #{number}"
+
+    lines = [f"{header}: {title}" if title else header]
+    if action:
+        lines.append(f"Action: {action}")
+    if html_url:
+        lines.append(f"URL: {html_url}")
+    if description:
+        lines.extend(["", "Description:", description[:MAX_BODY_CHARS]])
+
+    diff = _inline_diff(body, pull_request)
+    if diff:
+        lines.extend(["", "Diff:", diff[:MAX_DIFF_CHARS]])
+
+    peer_id = None
+    if repo and number is not None:
+        peer_id = f"{repo}#{number}"
+    elif number is not None:
+        peer_id = f"pr#{number}"
+    return "\n".join(lines), peer_id, _github_actor(body)
+
+
+def _repo_full_name(body: dict[str, Any]) -> str | None:
+    repo = body.get("repository")
+    if isinstance(repo, dict) and isinstance(repo.get("full_name"), str):
+        return repo["full_name"] or None
+    return None
+
+
+def _github_actor(body: dict[str, Any]) -> str:
+    sender = body.get("sender")
+    if isinstance(sender, dict) and isinstance(sender.get("login"), str) and sender["login"]:
+        return sender["login"]
+    return "webhook"
+
+
+def _inline_diff(body: dict[str, Any], pull_request: dict[str, Any]) -> str | None:
+    for source in (body, pull_request):
+        diff = source.get("diff")
+        if isinstance(diff, str) and diff.strip():
+            return diff.strip()
+    return None
+
+
+# ------------------------------------------------------------------- params resolution
+
+
+async def _default_fetch(url: str, headers: dict[str, str]) -> dict[str, Any]:
+    """GET a params source over HTTP. Imported lazily so callers that only pass inline
+    params carry no httpx dependency."""
+    import httpx
+
+    request_headers = {"accept": "application/json", **headers}
+    try:
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(url, headers=request_headers)
+            response.raise_for_status()
+            return response.json()
+    except httpx.HTTPError as exc:
+        raise WebhookError(f"params source request failed: {exc}") from exc
+    except ValueError as exc:  # json.JSONDecodeError subclasses ValueError
+        raise WebhookError(f"params source returned invalid JSON: {exc}") from exc
+
+
+async def resolve_remote_params(
+    url: str,
+    headers: dict[str, str] | None = None,
+    *,
+    fetch: ParamsFetcher | None = None,
+) -> tuple[dict[str, Any], dict[str, str]]:
+    """Fetch params (+ optional task_metadata) from a config-resolve URL.
+
+    Response shape (lenient)::
+
+        {"params": {...}, "task_metadata": {...}}
+
+    A bare object with no ``params`` key is treated as the params dict itself (minus a
+    top-level ``task_metadata``, which is returned separately for stamping).
+    """
+    do_fetch = fetch or (lambda u: _default_fetch(u, headers or {}))
+    payload = await do_fetch(url)
+    if not isinstance(payload, dict):
+        raise WebhookError("params source returned a non-object response")
+
+    metadata_raw = payload.get("task_metadata")
+    task_metadata = {str(k): str(v) for k, v in metadata_raw.items()} if isinstance(metadata_raw, dict) else {}
+    params = payload.get("params")
+    if not isinstance(params, dict):
+        params = {k: v for k, v in payload.items() if k != "task_metadata"}
+    return params, task_metadata
+
+
+# ------------------------------------------------------------------------- dispatch
+
+
+def _agent_reply_text(messages: object) -> str | None:
+    """Join agent-authored text from a message list (sync result or polled stream)."""
+    if not isinstance(messages, list):
+        return None
+    parts = []
+    for message in messages:
+        content = getattr(message, "content", None)
+        if (
+            content is not None
+            and getattr(content, "author", None) == "agent"
+            and getattr(content, "type", None) == "text"
+        ):
+            text = (getattr(content, "content", "") or "").strip()
+            if text:
+                parts.append(text)
+    return "\n\n".join(parts) if parts else None
+
+
+async def handle_webhook(
+    *,
+    agent_name: str,
+    payload: dict[str, Any],
+    acp_type: Literal["sync", "async"] = "sync",
+    shaper: Literal["generic", "github_pr"] = "generic",
+    channel: str | None = None,
+    params: dict[str, Any] | None = None,
+    params_source: str | None = None,
+    params_source_headers: dict[str, str] | None = None,
+    peer_id: str | None = None,
+    extra_task_metadata: dict[str, str] | None = None,
+    wait: bool = False,
+    fetch: ParamsFetcher | None = None,
+) -> WebhookResult:
+    """Drive an agent turn from a webhook payload, agent-side, via the ADK client.
+
+    - Shapes the payload (generic or GitHub PR) into a prompt + conversation scope.
+    - Resolves task params: inline ``params``, or fetched from ``params_source``
+      (config-by-id). The platform never interprets params — they're forwarded to the
+      agent as ``task/create`` params.
+    - Get-or-creates a task keyed on a stable session key, so repeat events fold in.
+    - Sends the turn (sync → message/send returns the reply inline; async → event/send,
+      with optional ``wait`` to poll for the reply).
+    """
+    channel = channel or shaper
+    if shaper == "github_pr":
+        text, derived_peer, sender = shape_github_pr(payload)
+        peer_id = peer_id or derived_peer
+    else:
+        text, sender = render_generic(payload), "webhook"
+
+    task_metadata: dict[str, str] = {"channel": channel, "sender_id": sender}
+    if peer_id:
+        task_metadata["peer_id"] = peer_id
+
+    resolved_params = dict(params) if params else {}
+    if params_source:
+        resolved_params, source_metadata = await resolve_remote_params(
+            params_source, params_source_headers, fetch=fetch
+        )
+        # Source metadata + caller extras never override the canonical fields above.
+        for key, value in {**source_metadata, **(extra_task_metadata or {})}.items():
+            task_metadata.setdefault(key, str(value))
+    elif extra_task_metadata:
+        for key, value in extra_task_metadata.items():
+            task_metadata.setdefault(key, str(value))
+
+    name = session_key(agent_name, channel, peer_id or "")
+    task = await adk.acp.create_task(
+        name=name,
+        agent_name=agent_name,
+        params=resolved_params or None,
+        request={"task_metadata": task_metadata} if task_metadata else None,
+    )
+
+    content = TextContent(author="user", content=text, format="markdown")
+
+    if acp_type == "sync":
+        messages = await adk.acp.send_message(task_id=task.id, agent_name=agent_name, content=content)
+        return WebhookResult(task_id=task.id, reply=_agent_reply_text(messages), task_metadata=task_metadata)
+
+    await adk.acp.send_event(task_id=task.id, agent_name=agent_name, content=content)
+    reply = await _await_reply(task.id) if wait else None
+    return WebhookResult(task_id=task.id, reply=reply, task_metadata=task_metadata)
+
+
+async def _await_reply(
+    task_id: str, *, timeout_s: float = 120.0, interval_s: float = 2.0, quiescence_s: float = 6.0
+) -> str | None:
+    """Poll a task's messages for the agent's reply until it settles or times out."""
+    import asyncio
+
+    waited = 0.0
+    last: str | None = None
+    stable_for = 0.0
+    while waited < timeout_s:
+        await asyncio.sleep(interval_s)
+        waited += interval_s
+        messages = await adk.messages.list(task_id=task_id)
+        text = _agent_reply_text(messages)
+        if text and text == last:
+            stable_for += interval_s
+            if stable_for >= quiescence_s:
+                return text
+        elif text:
+            last, stable_for = text, 0.0
+    return last