"""Support email refund workflow agent. Receives support email, checks Stripe refund eligibility with caller-provided credentials, drafts a customer reply, requires explicit approval before any refund mutation, and posts redacted audit summaries to Slack. """ from __future__ import annotations import asyncio import hashlib import json import re import time from email.utils import parseaddr from typing import Any import httpx from pydantic import BaseModel, Field import a2a_pack as a2a from a2a_pack import ( A2AAgent, ConsumerSetup, ConsumerSetupField, EgressPolicy, NoAuth, Pricing, Resources, RunContext, WorkspaceAccess, WorkspaceMode, ) LEDGER_PATH = "outputs/support-refund-audit-ledger.json" MAX_EMAIL_BODY_CHARS = 12_000 MAX_REFUND_AMOUNT_CENTS = 100_000 DEFAULT_CURRENCY = "usd" HTTP_TIMEOUT = httpx.Timeout(12.0, connect=4.0, read=8.0, write=8.0) class ReceivesSupportEmailChecks1919313Config(BaseModel): max_auto_eligible_refund_cents: int = Field( default=50_000, ge=1, le=MAX_REFUND_AMOUNT_CENTS, description="Maximum refund amount the agent may mark eligible for approval.", ) allowed_refund_currencies: list[str] = Field( default_factory=lambda: [DEFAULT_CURRENCY], description="Lowercase ISO currencies that may be proposed for refunds.", ) class ReceivesSupportEmailChecks1919313(A2AAgent[ReceivesSupportEmailChecks1919313Config, NoAuth]): name = "receives-support-email-checks-1-919313" description = ( "Receives support email, checks Stripe refund eligibility with caller-provided " "credentials, drafts the customer reply, requires explicit approval before " "executing refunds, and posts redacted audit summaries to Slack." ) version = "0.1.0" config_model = ReceivesSupportEmailChecks1919313Config auth_model = NoAuth consumer_setup = ConsumerSetup.from_fields( ConsumerSetupField.secret( "STRIPE_SECRET_KEY", label="Stripe secret key", description="Caller-provided Stripe API key used only server-side to inspect and create refunds.", input_type="password", ), ConsumerSetupField.secret( "SLACK_BOT_TOKEN", label="Slack bot token", description="Caller-provided Slack token used only server-side for audit summaries.", input_type="password", ), ConsumerSetupField.config( "SLACK_CHANNEL_ID", label="Slack audit channel ID", description="Slack channel where refund audit summaries should be posted.", input_type="text", ), ConsumerSetupField.config( "STRIPE_BASE_URL", label="Stripe API base URL", description="Optional test override. Defaults to https://api.stripe.com.", required=False, input_type="url", ), ConsumerSetupField.config( "SLACK_BASE_URL", label="Slack API base URL", description="Optional test override. Defaults to https://slack.com/api.", required=False, input_type="url", ), ) resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=300) workspace_access = WorkspaceAccess.dynamic( max_files=8, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), require_reason=False, ) egress = EgressPolicy(allow_hosts=("api.stripe.com", "slack.com")) tools_used = ("email", "stripe", "slack", "httpx") pricing = Pricing( price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic support refund workflow. Provider API usage is paid through caller-provided Stripe and Slack credentials.", ) @a2a.tool( description="Handle one inbound support email: dedupe, check Stripe refund eligibility, draft reply, and post a Slack audit summary without executing a refund.", timeout_seconds=120, idempotent=True, max_retries=0, cost_class="external-api", ) async def process_support_email( self, ctx: RunContext[NoAuth], sender_email: str, subject: str, body: str, message_id: str, stripe_payment_intent_id: str, requested_refund_amount_cents: int, currency: str = DEFAULT_CURRENCY, customer_name: str = "customer", ) -> dict[str, Any]: """Preview a refund workflow from bounded, typed email fields.""" email = { "sender": sender_email, "subject": subject, "body": body[:MAX_EMAIL_BODY_CHARS], "message_id": message_id, "references": [], "attachments": [], "date": None, } return await self._process_email_payload( ctx, email=email, stripe_payment_intent_id=stripe_payment_intent_id, requested_refund_amount_cents=requested_refund_amount_cents, currency=currency, customer_name=customer_name, ) @a2a.tool( description="Execute a previously proposed Stripe refund only when the caller supplies explicit approval bound to the exact action digest.", timeout_seconds=120, idempotent=True, max_retries=0, cost_class="payment-mutation", ) async def execute_approved_refund( self, ctx: RunContext[NoAuth], action_digest: str, approval_token: str, stripe_payment_intent_id: str, amount_cents: int, currency: str = DEFAULT_CURRENCY, customer_email: str = "", reason: str = "requested_by_customer", ) -> dict[str, Any]: """Execute one approved refund with a Stripe idempotency key.""" setup = _read_setup(ctx) if setup["status"] != "ready": return setup normalized = _normalize_refund_action( payment_intent_id=stripe_payment_intent_id, amount_cents=amount_cents, currency=currency, customer_email=customer_email, reason=reason, ) expected_digest = _action_digest(normalized) if expected_digest != action_digest: return { "status": "declined", "decline_code": "tampered_action", "message": "The approval digest does not match the requested refund action.", "expected_action_digest": expected_digest, "provided_action_digest": action_digest, } expected_token = _approval_phrase(action_digest) if approval_token != expected_token: return { "status": "approval_required", "approval_required": True, "action_digest": action_digest, "approval_instructions": f"To execute, submit approval_token exactly as: {expected_token}", } ledger = await _load_ledger(ctx) idempotency_key = _idempotency_key("refund", action_digest) if idempotency_key in ledger.get("refunds", {}): previous = ledger["refunds"][idempotency_key] return { "status": "duplicate_execution", "executed": previous.get("executed", False), "idempotency_key": idempotency_key, "provider_refund_id": previous.get("provider_refund_id"), "audit": previous.get("audit"), } await ctx.emit_progress("Creating approved Stripe refund with idempotency key") stripe_result = await _create_stripe_refund(setup, normalized, idempotency_key) now = _now() audit = { "event": "refund_execution", "status": stripe_result["status"], "action_digest": action_digest, "idempotency_key": idempotency_key, "payment_intent_id": stripe_payment_intent_id, "amount_cents": amount_cents, "currency": currency.lower(), "customer_email_hash": _hash_customer(customer_email), "timestamp": now, } if stripe_result["status"] == "executed": audit["provider_refund_id"] = stripe_result.get("provider_refund_id") slack_result = await _post_slack_audit(setup, audit) audit["slack_status"] = slack_result["status"] ledger.setdefault("refunds", {})[idempotency_key] = { "executed": stripe_result["status"] == "executed", "provider_refund_id": stripe_result.get("provider_refund_id"), "audit": audit, "created_at": now, } await _save_ledger(ctx, ledger) if stripe_result["status"] == "timeout": return { "status": "needs_reconciliation", "executed": False, "idempotency_key": idempotency_key, "action_digest": action_digest, "message": "Stripe timed out. Reconcile by checking Stripe for the idempotency key before retrying.", "audit": audit, "slack": slack_result, } if stripe_result["status"] != "executed": return { "status": "declined", "executed": False, "decline_code": stripe_result.get("error_type", "provider_error"), "idempotency_key": idempotency_key, "action_digest": action_digest, "audit": audit, "slack": slack_result, } return { "status": "executed", "executed": True, "provider_refund_id": stripe_result.get("provider_refund_id"), "idempotency_key": idempotency_key, "action_digest": action_digest, "audit": audit, "slack": slack_result, } @a2a.tool( description="Inbound email handler for the agent mailbox. It deduplicates and returns a draft reply; refund execution still requires a separate explicit approval call.", on_email=True, timeout_seconds=120, idempotent=True, max_retries=0, cost_class="external-api", ) async def receive_support_email( self, ctx: RunContext[NoAuth], email: dict[str, Any], ) -> dict[str, str] | None: parsed = _extract_refund_request(email) if not parsed.get("stripe_payment_intent_id"): return { "subject": _reply_subject(email.get("subject", "Support request")), "body": ( "Thanks for contacting support. We could not identify a Stripe payment intent " "in your message, so a teammate will review this manually." ), } result = await self._process_email_payload( ctx, email=email, stripe_payment_intent_id=parsed["stripe_payment_intent_id"], requested_refund_amount_cents=int(parsed.get("amount_cents") or 0), currency=str(parsed.get("currency") or DEFAULT_CURRENCY), customer_name=parsed.get("customer_name") or "customer", ) draft = result.get("draft_reply") or {} body = str(draft.get("body") or "Thanks for contacting support. We are reviewing your refund request.") return {"subject": str(draft.get("subject") or _reply_subject(email.get("subject", "Support request"))), "body": body} async def _process_email_payload( self, ctx: RunContext[NoAuth], *, email: dict[str, Any], stripe_payment_intent_id: str, requested_refund_amount_cents: int, currency: str, customer_name: str, ) -> dict[str, Any]: setup = _read_setup(ctx) if setup["status"] != "ready": return setup validation = _validate_refund_request( stripe_payment_intent_id=stripe_payment_intent_id, amount_cents=requested_refund_amount_cents, currency=currency, allowed_currencies=self.config.allowed_refund_currencies, max_refund_cents=self.config.max_auto_eligible_refund_cents, ) if validation["status"] != "valid": return validation message_id = str(email.get("message_id") or "").strip() sender = _sender_email(str(email.get("sender") or "")) email_key = _email_key(message_id, sender, stripe_payment_intent_id) ledger = await _load_ledger(ctx) if email_key in ledger.get("emails", {}): previous = ledger["emails"][email_key] return { "status": "duplicate_email", "duplicate": True, "message_id": message_id, "previous_action_digest": previous.get("action_digest"), "draft_reply": previous.get("draft_reply"), "audit": previous.get("audit"), } await ctx.emit_progress("Checking Stripe payment intent refund eligibility") stripe_check = await _check_stripe_payment_intent(setup, stripe_payment_intent_id) policy = _evaluate_refund_policy( stripe_check=stripe_check, requested_amount_cents=requested_refund_amount_cents, currency=currency, max_refund_cents=self.config.max_auto_eligible_refund_cents, ) normalized_action = _normalize_refund_action( payment_intent_id=stripe_payment_intent_id, amount_cents=requested_refund_amount_cents, currency=currency, customer_email=sender, reason="requested_by_customer", ) action_digest = _action_digest(normalized_action) draft_reply = _draft_customer_reply( sender=sender, subject=str(email.get("subject") or "Support request"), customer_name=customer_name, policy=policy, amount_cents=requested_refund_amount_cents, currency=currency, action_digest=action_digest, ) audit = { "event": "refund_preview", "status": policy["status"], "message_id_hash": _stable_hash(message_id), "customer_email_hash": _hash_customer(sender), "payment_intent_id": stripe_payment_intent_id, "amount_cents": requested_refund_amount_cents, "currency": currency.lower(), "action_digest": action_digest, "approval_required": policy.get("approval_required", True), "timestamp": _now(), } slack_result = await _post_slack_audit(setup, audit) audit["slack_status"] = slack_result["status"] ledger.setdefault("emails", {})[email_key] = { "action_digest": action_digest, "draft_reply": draft_reply, "audit": audit, "created_at": _now(), } await _save_ledger(ctx, ledger) return { "status": policy["status"], "duplicate": False, "approval_required": True, "executed": False, "action_digest": action_digest, "idempotency_key_preview": _idempotency_key("refund", action_digest), "approval_instructions": ( "Review the draft and policy result. To execute the refund, call " "execute_approved_refund with this action_digest and approval_token " f"exactly: {_approval_phrase(action_digest)}" ), "policy": policy, "draft_reply": draft_reply, "audit": audit, "slack": slack_result, } def _read_setup(ctx: RunContext[Any]) -> dict[str, Any]: missing: list[str] = [] values: dict[str, str] = {} for name in ("STRIPE_SECRET_KEY", "SLACK_BOT_TOKEN"): try: values[name] = ctx.consumer_secret(name) except Exception: missing.append(name) slack_channel = str(ctx.consumer_config("SLACK_CHANNEL_ID", "") or "").strip() if not slack_channel: missing.append("SLACK_CHANNEL_ID") if missing: return { "status": "setup_required", "missing": sorted(set(missing)), "message": "Configure caller-provided Stripe and Slack setup before running this workflow.", } return { "status": "ready", "stripe_key": values["STRIPE_SECRET_KEY"], "slack_token": values["SLACK_BOT_TOKEN"], "slack_channel": slack_channel, "stripe_base_url": _clean_base_url(ctx.consumer_config("STRIPE_BASE_URL", "https://api.stripe.com")), "slack_base_url": _clean_base_url(ctx.consumer_config("SLACK_BASE_URL", "https://slack.com/api")), } def _clean_base_url(value: Any) -> str: text = str(value or "").strip().rstrip("/") return text or "https://api.stripe.com" def _validate_refund_request( *, stripe_payment_intent_id: str, amount_cents: int, currency: str, allowed_currencies: list[str], max_refund_cents: int, ) -> dict[str, Any]: if not re.fullmatch(r"pi_[A-Za-z0-9_]+", stripe_payment_intent_id or ""): return {"status": "declined", "decline_code": "invalid_payment_intent", "executed": False} currency_clean = (currency or "").lower().strip() if currency_clean not in {c.lower() for c in allowed_currencies}: return {"status": "declined", "decline_code": "unsupported_currency", "currency": currency_clean, "executed": False} if amount_cents <= 0 or amount_cents > min(max_refund_cents, MAX_REFUND_AMOUNT_CENTS): return {"status": "declined", "decline_code": "amount_out_of_policy", "max_refund_cents": max_refund_cents, "executed": False} return {"status": "valid"} async def _check_stripe_payment_intent(setup: dict[str, Any], payment_intent_id: str) -> dict[str, Any]: url = f"{setup['stripe_base_url']}/v1/payment_intents/{payment_intent_id}" try: async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client: resp = await client.get( url, headers={"Authorization": f"Bearer {setup['stripe_key']}"}, params={"expand[]": "latest_charge"}, ) except (httpx.TimeoutException, asyncio.TimeoutError): return {"status": "timeout", "error_type": "provider_timeout"} except httpx.HTTPError as exc: return {"status": "provider_error", "error_type": type(exc).__name__} if resp.status_code in {401, 403}: return {"status": "auth_error", "error_type": "stripe_auth"} if resp.status_code == 404: return {"status": "not_found", "error_type": "payment_intent_not_found"} if resp.status_code == 429: return {"status": "rate_limited", "error_type": "stripe_rate_limit"} if resp.status_code >= 400: return {"status": "provider_error", "error_type": "stripe_error", "status_code": resp.status_code} try: data = resp.json() except ValueError: return {"status": "provider_error", "error_type": "malformed_json"} return { "status": "ok", "payment_intent_status": data.get("status"), "amount_received": int(data.get("amount_received") or data.get("amount") or 0), "amount_refunded": _amount_refunded(data), "currency": str(data.get("currency") or "").lower(), "livemode": bool(data.get("livemode")), } def _amount_refunded(payment_intent: dict[str, Any]) -> int: charge = payment_intent.get("latest_charge") if isinstance(charge, dict): return int(charge.get("amount_refunded") or 0) charges = payment_intent.get("charges") if isinstance(charges, dict): data = charges.get("data") or [] if data and isinstance(data[0], dict): return int(data[0].get("amount_refunded") or 0) return 0 def _evaluate_refund_policy( *, stripe_check: dict[str, Any], requested_amount_cents: int, currency: str, max_refund_cents: int, ) -> dict[str, Any]: if stripe_check["status"] == "timeout": return {"status": "needs_reconciliation", "approval_required": True, "reason": "Stripe timed out; reconcile before retrying."} if stripe_check["status"] != "ok": return {"status": "declined", "approval_required": True, "reason": stripe_check.get("error_type", "stripe_lookup_failed")} if stripe_check.get("currency") != currency.lower(): return {"status": "declined", "approval_required": True, "reason": "Currency mismatch."} if stripe_check.get("payment_intent_status") not in {"succeeded", "requires_capture"}: return {"status": "declined", "approval_required": True, "reason": "Payment intent is not in a refundable state."} refundable = max(0, int(stripe_check.get("amount_received") or 0) - int(stripe_check.get("amount_refunded") or 0)) if requested_amount_cents > refundable: return {"status": "declined", "approval_required": True, "reason": "Requested refund exceeds remaining refundable amount.", "refundable_amount_cents": refundable} if requested_amount_cents > max_refund_cents: return {"status": "declined", "approval_required": True, "reason": "Requested refund exceeds configured policy limit."} return {"status": "proposed", "approval_required": True, "refundable_amount_cents": refundable, "reason": "Eligible for an approved refund."} async def _create_stripe_refund(setup: dict[str, Any], action: dict[str, Any], idempotency_key: str) -> dict[str, Any]: try: async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client: resp = await client.post( f"{setup['stripe_base_url']}/v1/refunds", headers={ "Authorization": f"Bearer {setup['stripe_key']}", "Idempotency-Key": idempotency_key, }, data={ "payment_intent": action["payment_intent_id"], "amount": str(action["amount_cents"]), "reason": action["reason"], "metadata[action_digest]": _action_digest(action), }, ) except (httpx.TimeoutException, asyncio.TimeoutError): return {"status": "timeout", "error_type": "provider_timeout"} except httpx.HTTPError as exc: return {"status": "provider_error", "error_type": type(exc).__name__} if resp.status_code in {401, 403}: return {"status": "provider_error", "error_type": "stripe_auth"} if resp.status_code == 429: return {"status": "provider_error", "error_type": "stripe_rate_limit"} if resp.status_code >= 400: return {"status": "provider_error", "error_type": "stripe_error", "status_code": resp.status_code} try: data = resp.json() except ValueError: return {"status": "provider_error", "error_type": "malformed_json"} return {"status": "executed", "provider_refund_id": data.get("id"), "provider_status": data.get("status")} async def _post_slack_audit(setup: dict[str, Any], audit: dict[str, Any]) -> dict[str, Any]: payload = { "channel": setup["slack_channel"], "text": _slack_audit_text(audit), "unfurl_links": False, "unfurl_media": False, } try: async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client: resp = await client.post( f"{setup['slack_base_url']}/chat.postMessage", headers={"Authorization": f"Bearer {setup['slack_token']}", "Content-Type": "application/json"}, json=payload, ) except (httpx.TimeoutException, asyncio.TimeoutError): return {"status": "timeout", "message": "Slack audit post timed out."} except httpx.HTTPError as exc: return {"status": "provider_error", "error_type": type(exc).__name__} if resp.status_code in {401, 403}: return {"status": "auth_error"} if resp.status_code == 429: return {"status": "rate_limited"} if resp.status_code >= 400: return {"status": "provider_error", "status_code": resp.status_code} try: data = resp.json() except ValueError: return {"status": "provider_error", "error_type": "malformed_json"} if data.get("ok") is not True: return {"status": "provider_error", "error_type": str(data.get("error") or "slack_error")} return {"status": "posted", "message_ts": data.get("ts"), "channel": data.get("channel")} def _slack_audit_text(audit: dict[str, Any]) -> str: return ( f"Support refund audit: {audit.get('event')} status={audit.get('status')} " f"pi={audit.get('payment_intent_id')} amount={audit.get('amount_cents')} {audit.get('currency')} " f"digest={audit.get('action_digest')} customer_hash={audit.get('customer_email_hash')}" ) def _draft_customer_reply(*, sender: str, subject: str, customer_name: str, policy: dict[str, Any], amount_cents: int, currency: str, action_digest: str) -> dict[str, str]: dollars = f"{amount_cents / 100:.2f} {currency.upper()}" greeting = f"Hi {customer_name.strip() or 'there'}," if policy["status"] == "proposed": body = ( f"{greeting}\n\nThanks for contacting support. We found your payment and your requested refund of {dollars} " "appears eligible under our refund policy. For your protection, no refund has been issued yet; " "a support teammate must approve the exact refund action first.\n\n" f"Reference: {action_digest}\n\nWe will follow up once the review is complete." ) elif policy["status"] == "needs_reconciliation": body = ( f"{greeting}\n\nThanks for contacting support. We are reviewing your refund request, but our payment provider " "did not respond in time. A teammate will reconcile the payment record before taking any action." ) else: body = ( f"{greeting}\n\nThanks for contacting support. We reviewed the refund request and cannot automatically mark it " f"eligible because: {policy.get('reason', 'it requires manual review')}. A teammate will review it manually." ) return {"to": sender, "subject": _reply_subject(subject), "body": body} async def _load_ledger(ctx: RunContext[Any]) -> dict[str, Any]: try: view = await ctx.workspace.open_view( purpose="read support refund audit ledger", hints=[LEDGER_PATH], max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="Deduplicate emails and refund executions.", ) for f in view.files: if f.path == LEDGER_PATH: raw = await view.read(LEDGER_PATH) return json.loads(raw.decode("utf-8")) except Exception: pass return {"emails": {}, "refunds": {}} async def _save_ledger(ctx: RunContext[Any], ledger: dict[str, Any]) -> None: data = json.dumps(ledger, sort_keys=True, indent=2).encode("utf-8") try: grant = await ctx.workspace.request_access( files=[LEDGER_PATH], mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="Persist support email and refund idempotency ledger.", purpose="write support refund audit ledger", ) view = await ctx.workspace.open_view( purpose="write support refund audit ledger", hints=[LEDGER_PATH], max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason=f"Use grant {grant.grant_id} to write audit ledger.", ) await view.write(LEDGER_PATH, data) except Exception: ref = await ctx.write_artifact("support-refund-audit-ledger.json", data, "application/json") await ctx.emit_artifact(ref) def _normalize_refund_action(*, payment_intent_id: str, amount_cents: int, currency: str, customer_email: str, reason: str) -> dict[str, Any]: return { "payment_intent_id": payment_intent_id.strip(), "amount_cents": int(amount_cents), "currency": currency.lower().strip(), "customer_email_hash": _hash_customer(customer_email), "reason": reason.strip() or "requested_by_customer", } def _action_digest(action: dict[str, Any]) -> str: return hashlib.sha256(json.dumps(action, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest() def _approval_phrase(action_digest: str) -> str: return f"APPROVE-REFUND-{action_digest[:16]}" def _idempotency_key(prefix: str, digest: str) -> str: return f"a2a-{prefix}-{digest[:32]}" def _email_key(message_id: str, sender: str, payment_intent_id: str) -> str: return _stable_hash("|".join([message_id.strip().lower(), sender.strip().lower(), payment_intent_id.strip()])) def _hash_customer(value: str) -> str: return _stable_hash((value or "").strip().lower())[:16] def _stable_hash(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def _sender_email(value: str) -> str: parsed = parseaddr(value)[1] return parsed or value.strip() def _reply_subject(subject: str) -> str: text = (subject or "Support request").strip() return text if text.lower().startswith("re:") else f"Re: {text}" def _extract_refund_request(email: dict[str, Any]) -> dict[str, Any]: text = f"{email.get('subject', '')}\n{email.get('body', '')}"[:MAX_EMAIL_BODY_CHARS] pi_match = re.search(r"\b(pi_[A-Za-z0-9_]+)\b", text) amount_match = re.search(r"(?:\$|USD\s*)?(\d+(?:\.\d{1,2})?)\s*(usd|USD)?", text) cents = int(float(amount_match.group(1)) * 100) if amount_match else 0 sender = _sender_email(str(email.get("sender") or "")) name = parseaddr(str(email.get("sender") or ""))[0] or "customer" return { "stripe_payment_intent_id": pi_match.group(1) if pi_match else "", "amount_cents": cents, "currency": "usd", "customer_email": sender, "customer_name": name, } def _now() -> int: return int(time.time())