| 1 | """ |
| 2 | Channel-specific delivery helpers for the notification dispatcher. |
| 3 | |
| 4 | Each helper is async, returns a tuple shape that includes status, |
| 5 | error_message, latency_ms, and optional provider-specific extras. None |
| 6 | raise — failures are reported via the tuple so the caller can log them |
| 7 | in a single shape. This keeps the dispatch loop's try/except surface |
| 8 | trivial. |
| 9 | |
| 10 | Channels: |
| 11 | - shuffle : POST to https://shuffler.io/api/v1/apps/{id}/mcp using |
| 12 | the deployment's Bearer (from the Shuffle connector) |
| 13 | + customer's Org-Id header. Fire-and-record — we |
| 14 | capture Shuffle's execution_id but don't poll for the |
| 15 | downstream provider's terminal state. Email, Slack, |
| 16 | Teams, etc. all flow through this single channel via |
| 17 | Shuffle's catalog of authenticated apps. |
| 18 | """ |
| 19 | |
| 20 | from __future__ import annotations |
| 21 | |
| 22 | import time |
| 23 | from typing import Any |
| 24 | from typing import Dict |
| 25 | from typing import Optional |
| 26 | from typing import Tuple |
| 27 | |
| 28 | import httpx |
| 29 | from loguru import logger |
| 30 | |
| 31 | # Shuffle's dispatcher returns a 4-tuple — status, error_message, |
| 32 | # latency_ms, and the kickoff execution_id (None if Shuffle didn't |
| 33 | # return one before the call failed). |
| 34 | ShuffleDispatchResult = Tuple[str, Optional[str], int, Optional[str]] |
| 35 | |
| 36 | |
| 37 | # Hard cap on Shuffle's MCP kickoff. Usually <1s but cold-start paths |
| 38 | # (first call against an org, or Shuffle's backend warming) can spike. |
| 39 | # 30s leaves enough headroom for those without letting a stuck request |
| 40 | # stall the dispatch loop indefinitely. |
| 41 | _SHUFFLE_TIMEOUT_S = 30.0 |
| 42 | |
| 43 | |
| 44 | # --------------------------------------------------------------------------- |
| 45 | # Shuffle dispatcher (Phase 2) |
| 46 | # --------------------------------------------------------------------------- |
| 47 | |
| 48 | |
| 49 | def _shuffle_headers(api_key: str, org_id: str) -> Dict[str, str]: |
| 50 | """Bearer auth + Org-Id scope. |
| 51 | |
| 52 | The deployment's Shuffle API key (admin-scoped, lives in CoPilot's |
| 53 | Shuffle connector row) plus the customer's per-integration Org-Id |
| 54 | is what scopes a dispatch to the right org. The MCP server in |
| 55 | shuffle-mcp-server uses the same pair. |
| 56 | """ |
| 57 | return { |
| 58 | "Authorization": f"Bearer {api_key}", |
| 59 | "Org-Id": org_id, |
| 60 | "Content-Type": "application/json", |
| 61 | "Accept": "application/json", |
| 62 | } |
| 63 | |
| 64 | |
| 65 | async def dispatch_shuffle( |
| 66 | *, |
| 67 | base_url: str, |
| 68 | api_key: str, |
| 69 | org_id: str, |
| 70 | app_id: str, |
| 71 | input_text: str, |
| 72 | environment: str = "Shuffle", |
| 73 | ) -> ShuffleDispatchResult: |
| 74 | """Kick off a Shuffle AI Agent run for one app. |
| 75 | |
| 76 | Wraps `POST {base_url}/api/v1/apps/{app_id}/mcp`. Shuffle's response |
| 77 | contains an `execution_id` + `authorization` — the actual downstream |
| 78 | delivery (Slack message, email send, etc.) happens asynchronously |
| 79 | inside Shuffle. Phase 2 is fire-and-record: we treat HTTP 200 from |
| 80 | Shuffle as `sent` and stash the execution_id for forensic lookups, |
| 81 | but we do NOT poll for terminal state. Phase 4 may add an optional |
| 82 | poll-with-timeout mode for high-criticality routes. |
| 83 | |
| 84 | Returns (status, error_message, latency_ms, execution_id_or_None). |
| 85 | """ |
| 86 | url = f"{base_url.rstrip('/')}/api/v1/apps/{app_id}/mcp" |
| 87 | body: Dict[str, Any] = { |
| 88 | "jsonrpc": "2.0", |
| 89 | "id": "1", |
| 90 | "method": "tools/call", |
| 91 | "params": { |
| 92 | "tool_id": app_id, |
| 93 | "tool_name": app_id, |
| 94 | "input": {"text": input_text}, |
| 95 | "environment": environment, |
| 96 | }, |
| 97 | } |
| 98 | logger.info(f"Dispatching payload to Shuffle body: {body}") |
| 99 | started = time.monotonic() |
| 100 | try: |
| 101 | async with httpx.AsyncClient(timeout=_SHUFFLE_TIMEOUT_S, http2=True) as client: |
| 102 | response = await client.post(url, headers=_shuffle_headers(api_key, org_id), json=body) |
| 103 | latency_ms = int((time.monotonic() - started) * 1000) |
| 104 | |
| 105 | if response.status_code in (401, 403): |
| 106 | return ( |
| 107 | "failed", |
| 108 | f"Shuffle authentication failed ({response.status_code}): {response.text[:200]}", |
| 109 | latency_ms, |
| 110 | None, |
| 111 | ) |
| 112 | if response.status_code >= 400: |
| 113 | return ( |
| 114 | "failed", |
| 115 | f"Shuffle returned {response.status_code}: {response.text[:200]}", |
| 116 | latency_ms, |
| 117 | None, |
| 118 | ) |
| 119 | |
| 120 | try: |
| 121 | data = response.json() |
| 122 | except ValueError: |
| 123 | return ("failed", "Shuffle returned non-JSON response", latency_ms, None) |
| 124 | |
| 125 | # Shuffle's success body shape: {success, execution_id, authorization, mode}. |
| 126 | # `execution_id` is what we stash for forensic correlation in |
| 127 | # the dispatch log; no polling. |
| 128 | execution_id = data.get("execution_id") if isinstance(data, dict) else None |
| 129 | if isinstance(data, dict) and data.get("success") is False: |
| 130 | return ( |
| 131 | "failed", |
| 132 | f"Shuffle reported failure: {data.get('reason') or data.get('error') or data}", |
| 133 | latency_ms, |
| 134 | execution_id, |
| 135 | ) |
| 136 | return ("sent", None, latency_ms, execution_id) |
| 137 | except Exception as e: # noqa: BLE001 |
| 138 | latency_ms = int((time.monotonic() - started) * 1000) |
| 139 | logger.warning(f"Shuffle dispatch failed: {e!r}") |
| 140 | return ("failed", f"{type(e).__name__}: {e}", latency_ms, None) |
| 141 | |
| 142 | |
| 143 | async def list_shuffle_apps( |
| 144 | *, |
| 145 | base_url: str, |
| 146 | api_key: str, |
| 147 | org_id: str, |
| 148 | ) -> Tuple[bool, list, Optional[str]]: |
| 149 | """Fetch the apps catalog the customer's Shuffle org has access to. |
| 150 | |
| 151 | Used by the route form's app picker so admins can pick from a list |
| 152 | instead of hand-typing UUIDs. Returns (ok, apps, error_message). |
| 153 | """ |
| 154 | url = f"{base_url.rstrip('/')}/api/v1/apps" |
| 155 | try: |
| 156 | async with httpx.AsyncClient(timeout=_SHUFFLE_TIMEOUT_S, http2=True) as client: |
| 157 | response = await client.get(url, headers=_shuffle_headers(api_key, org_id)) |
| 158 | if response.status_code in (401, 403): |
| 159 | return (False, [], f"Shuffle authentication failed ({response.status_code})") |
| 160 | if response.status_code >= 400: |
| 161 | return (False, [], f"Shuffle returned {response.status_code}: {response.text[:200]}") |
| 162 | try: |
| 163 | data = response.json() |
| 164 | except ValueError: |
| 165 | return (False, [], "Shuffle returned non-JSON response") |
| 166 | |
| 167 | # Shuffle returns the catalog as a list of app objects with at |
| 168 | # minimum {id, name, description}. We forward the minimal shape |
| 169 | # the UI needs and let the route form record (id, name) on submit. |
| 170 | if not isinstance(data, list): |
| 171 | return (False, [], f"Unexpected Shuffle response shape: {type(data).__name__}") |
| 172 | return (True, data, None) |
| 173 | except Exception as e: # noqa: BLE001 |
| 174 | logger.warning(f"Shuffle apps list failed: {e!r}") |
| 175 | return (False, [], f"{type(e).__name__}: {e}") |
| 176 | |
| 177 | |
| 178 | async def verify_shuffle_org( |
| 179 | *, |
| 180 | base_url: str, |
| 181 | api_key: str, |
| 182 | org_id: str, |
| 183 | ) -> Tuple[bool, Optional[int], Optional[str]]: |
| 184 | """Quick auth probe for an integration. Used by the 'Test connection' |
| 185 | button in the integration form. Returns (ok, app_count, error).""" |
| 186 | ok, apps, error = await list_shuffle_apps(base_url=base_url, api_key=api_key, org_id=org_id) |
| 187 | if not ok: |
| 188 | return (False, None, error) |
| 189 | return (True, len(apps), None) |
| 190 | |
| 191 | |
| 192 | async def list_shuffle_orgs( |
| 193 | *, |
| 194 | base_url: str, |
| 195 | api_key: str, |
| 196 | ) -> Tuple[bool, list, Optional[str]]: |
| 197 | """Fetch the orgs the deployment's admin Bearer can see. |
| 198 | |
| 199 | Used by the integration form so admins pick from a dropdown of real |
| 200 | orgs instead of pasting Org-Ids. Hits `GET /api/v1/orgs` — Shuffle |
| 201 | treats the admin key as having visibility into the parent org plus |
| 202 | any sub-orgs. We deliberately do NOT send an `Org-Id` header: the |
| 203 | request is unscoped so we get the full list back rather than a |
| 204 | single-org view. |
| 205 | |
| 206 | Returns (ok, orgs, error_message). Each org element is a dict with |
| 207 | at least `id` and `name`; the service layer trims it to the shape |
| 208 | the frontend dropdown needs. |
| 209 | """ |
| 210 | url = f"{base_url.rstrip('/')}/api/v1/orgs" |
| 211 | headers = { |
| 212 | "Authorization": f"Bearer {api_key}", |
| 213 | "Accept": "application/json", |
| 214 | } |
| 215 | try: |
| 216 | async with httpx.AsyncClient(timeout=_SHUFFLE_TIMEOUT_S, http2=True) as client: |
| 217 | response = await client.get(url, headers=headers) |
| 218 | if response.status_code in (401, 403): |
| 219 | return (False, [], f"Shuffle authentication failed ({response.status_code})") |
| 220 | if response.status_code >= 400: |
| 221 | return (False, [], f"Shuffle returned {response.status_code}: {response.text[:200]}") |
| 222 | try: |
| 223 | data = response.json() |
| 224 | except ValueError: |
| 225 | return (False, [], "Shuffle returned non-JSON response") |
| 226 | |
| 227 | # Shuffle's `/api/v1/orgs` historically returns a flat list of |
| 228 | # org objects; some installs wrap it in `{"orgs": [...]}`. Tolerate |
| 229 | # both shapes so we don't break on an upstream change. |
| 230 | if isinstance(data, dict) and "orgs" in data: |
| 231 | data = data.get("orgs") or [] |
| 232 | if not isinstance(data, list): |
| 233 | return (False, [], f"Unexpected Shuffle response shape: {type(data).__name__}") |
| 234 | return (True, data, None) |
| 235 | except Exception as e: # noqa: BLE001 |
| 236 | logger.warning(f"Shuffle orgs list failed: {e!r}") |
| 237 | return (False, [], f"{type(e).__name__}: {e}") |