Approval delivery adapters

Hangar's approval gate holds a tools/call until a human decides. Where that decision is asked for, and how the answer comes back, is not core's business.

Core ships two channels — dashboard and noop — and resolves everything else from the mcp_hangar.approvals.delivery entry-point group. A vendor integration is a package you install, not a branch in the gateway. The reasoning is in ADR-016.

The gate only started holding calls in 2.1.0. Everything on this page describes delivery, which is downstream of a gate that until then was not reachable on any shipped build: no config key put a tool behind it, the service was never constructed, and the REST routes answered 500 (#678). If you wrote an adapter against an earlier release and never saw it fire, that is why — there was nothing to deliver. Put a tool behind the gate with approval_list first.

Putting a tool behind the gate

Delivery is only reached once a policy gates a tool. That is a tools: block:

mcp_servers:
  payments:
    mode: remote
    endpoint: https://payments.example.com/mcp
    tools:
      approval_list:
        - "refund_*"
      approval_timeout_seconds: 600
      approval_channel: slack       # the entry-point name of your adapter

approval_channel names the channel this policy's approvals are delivered on, so different servers can route to different adapters. An unknown channel degrades to noop with a warning: approvals still queue and stay resolvable over REST, but nobody is notified. Full key reference: Configuration → tools dual format.

Migrating from approvals.channel: slack? Core carried a built-in Slack channel through 1.x. It was removed in 2.0. Nothing breaks silently: the channel now logs approval_delivery_channel_unknown and degrades to noop, so approvals queue undelivered but stay resolvable over the REST API. Restore delivery by installing an adapter — the full reference implementation is below.

The two halves

An adapter has an outbound half and an inbound half, and the inbound half is the one that used to live in core.

Outbound — notify the approver. Implement send; that is the whole protocol.

Inbound — take the answer back. Your adapter terminates the vendor's webhook itself: it verifies the vendor's signature, maps the vendor identity onto a Hangar principal, and then calls Hangar's ordinary API with an ordinary token.

That inbound shape is the point of the design. Core's resolve endpoint used to branch on an X-Slack-Signature header and dispatch to a vendor verifier, which meant an unauthenticated caller chose which authentication mechanism ran. Now there is one authentication path and one authorized chokepoint; your adapter is just another authenticated client of it.

Slack ──webhook──▶ your adapter ──POST /approvals/{id}/resolve──▶ Hangar
                   verifies HMAC       Bearer <hangar token>
                   maps user→principal

Registering a channel

# pyproject.toml of your adapter package
[project.entry-points."mcp_hangar.approvals.delivery"]
slack = "my_hangar_slack:build_delivery"

The entry point resolves to a callable taking the channel's config block and returning anything satisfying the ApprovalDelivery protocol:

def build_delivery(config: dict):
    return SlackApprovalDelivery(
        webhook_url=config["webhook_url"],
        signing_secret=config["signing_secret"],
    )

Then in Hangar's config:

approvals:
  channel: slack          # matches the entry-point name
  slack:                  # passed to your factory as `config`
    webhook_url: ${SLACK_WEBHOOK_URL}
    signing_secret: ${SLACK_SIGNING_SECRET}

An adapter that fails to load, or to construct, degrades to noop with a warning rather than stopping the gateway — a missing notification channel should not be an outage.

Reference implementation: Slack

This is the code that used to ship in core, unchanged apart from the entry point, plus the inbound half.

Outbound

"""Slack approval delivery via incoming webhook."""

from typing import Any

from mcp_hangar.approvals.models import ApprovalRequest
from mcp_hangar.logging_config import get_logger

logger = get_logger(__name__)

MAX_ARG_DISPLAY = 500


def _sanitize_for_display(arguments: dict[str, Any]) -> str:
    """Render arguments for a human, truncated.

    Note what this does NOT do: redact. Hangar redacts secrets on the audit
    path, not here. If your channel is a shared room, treat the notification as
    readable by everyone in it and consider sending only the tool name.
    """
    import json

    text = json.dumps(arguments, indent=2, default=str)
    if len(text) > MAX_ARG_DISPLAY:
        text = text[:MAX_ARG_DISPLAY] + "\n… (truncated)"
    return text


def _build_slack_blocks(request: ApprovalRequest) -> list[dict[str, Any]]:
    """Block Kit payload with approve/deny buttons.

    `action_id` carries the approval id back to your webhook handler.
    """
    return [
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": (
                    f"*Approval required*\n"
                    f"*Tool:* `{request.tool_name}`\n"
                    f"*Server:* `{request.provider_id}`\n"
                    f"*Requested:* {request.requested_at.isoformat()}"
                ),
            },
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"```{_sanitize_for_display(request.arguments)}```"},
        },
        {
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Approve"},
                    "style": "primary",
                    "action_id": f"approve:{request.approval_id}",
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Deny"},
                    "style": "danger",
                    "action_id": f"deny:{request.approval_id}",
                },
            ],
        },
    ]


class SlackApprovalDelivery:
    """Sends approval notifications to Slack via incoming webhook."""

    def __init__(self, webhook_url: str, signing_secret: str) -> None:
        self._webhook_url = webhook_url
        self._signing_secret = signing_secret

    async def send(self, request: ApprovalRequest) -> None:
        """Notify. Logs and swallows errors -- delivery failure must not fail the call."""
        import httpx

        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                response = await client.post(
                    self._webhook_url,
                    json={"blocks": _build_slack_blocks(request)},
                )
                if response.status_code != 200:
                    logger.warning(
                        "slack_delivery_non_200",
                        approval_id=request.approval_id,
                        status=response.status_code,
                        body=response.text[:200],
                    )
                else:
                    logger.info(
                        "slack_approval_delivered",
                        approval_id=request.approval_id,
                        tool=request.tool_name,
                    )
        except Exception:  # noqa: BLE001
            logger.warning("slack_delivery_failed", approval_id=request.approval_id, exc_info=True)

Inbound

Your own HTTP endpoint, in your own process. This half used to be _handle_slack_callback inside Hangar's routes.

import hashlib
import hmac
import json
import time
from urllib.parse import parse_qs

import httpx

FRESHNESS_WINDOW_S = 300


def verify_slack_signature(signing_secret: str, headers, body: str) -> bool:
    """HMAC-SHA256 over `v0:{timestamp}:{body}`, with replay protection.

    The freshness window matters as much as the signature: without it a captured
    request stays valid forever. Compare in constant time.
    """
    timestamp = headers.get("x-slack-request-timestamp", "")
    signature = headers.get("x-slack-signature", "")
    if not timestamp or not signature:
        return False

    try:
        if abs(time.time() - int(timestamp)) > FRESHNESS_WINDOW_S:
            return False
    except ValueError:
        return False

    expected = (
        "v0="
        + hmac.new(
            signing_secret.encode(),
            f"v0:{timestamp}:{body}".encode(),
            hashlib.sha256,
        ).hexdigest()
    )
    return hmac.compare_digest(expected, signature)


async def handle_slack_callback(
    signing_secret: str,
    base_url: str,          # your deployment's Hangar URL -- you supply this
    headers,
    raw_body: str,
) -> None:
    if not verify_slack_signature(signing_secret, headers, raw_body):
        raise PermissionError("bad Slack signature")

    payload = json.loads(parse_qs(raw_body)["payload"][0])
    action_id = payload["actions"][0]["action_id"]        # "approve:<id>" | "deny:<id>"
    decision, approval_id = action_id.split(":", 1)
    slack_user = payload["user"]["id"]

    # The load-bearing line. Hangar's audit trail should name a Hangar
    # principal, not a vendor handle -- map it here, and refuse if you cannot.
    token = mint_hangar_token_for(slack_user)

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            f"{base_url}/api/approvals/{approval_id}/resolve",
            json={"decision": decision, "reason": f"via Slack by {slack_user}"},
            headers={"Authorization": f"Bearer {token}"},
        )
        response.raise_for_status()

mint_hangar_token_for is yours to implement, and it is where the security of this integration actually lives. It must establish that this Slack user corresponds to a Hangar principal holding approval:resolve — an OIDC exchange, a mapping table, whatever your identity story is. Do not mint a single shared service token for every approver: the audit trail would then record one identity for every decision, which is exactly the attribution problem this design removes.

Testing an adapter

Test the signature verification against known-good and tampered payloads, including a stale timestamp — that is the part core no longer checks for you.

Hangar's own test for the registry (test_delivery_registry.py) shows how a channel is resolved and what happens when one fails to load; it is a useful template for asserting that your entry point is discovered.

See also

• ADR-016 — why core has one authorization chokepoint and no vendors
• Authentication & Authorization — minting tokens and the approval:resolve permission
• REST API — the /approvals endpoints
• Configuration — approval_list, approval_timeout_seconds, approval_channel