diff --git a/autogpt_platform/backend/backend/blocks/stripe_link/EXPLORATION.md b/autogpt_platform/backend/backend/blocks/stripe_link/EXPLORATION.md new file mode 100644 index 0000000000..e4bbd0c872 --- /dev/null +++ b/autogpt_platform/backend/backend/blocks/stripe_link/EXPLORATION.md @@ -0,0 +1,210 @@ +# Stripe Link CLI Block — Auth Exploration + +## What is Stripe Link CLI? + +[`@stripe/link-cli`](https://github.com/stripe/link-cli) lets AI agents get secure, +one-time-use payment credentials from a user's **Link wallet** (Stripe's consumer +payment product). The core operations are: + +| Operation | Description | +|-----------|-------------| +| `auth login` | Authenticate the agent with a Link account | +| `payment-methods list` | List cards/bank accounts in the wallet | +| `spend-request create` | Request a one-time virtual card credential | +| `spend-request retrieve` | Get card details once user approves | +| `mpp pay` | Execute payment via Machine Payments Protocol (SPT) | + +## Link CLI's Auth Model + +Link CLI uses **OAuth 2.0 Device Code Grant** ([RFC 8628](https://tools.ietf.org/html/rfc8628)): + +``` +┌─────────┐ ┌──────────────────┐ ┌──────────┐ +│ AutoGPT │ │ login.link.com │ │ User's │ +│ Backend │ │ (Auth Server) │ │ Browser │ +└────┬────┘ └────────┬─────────┘ └────┬─────┘ + │ │ │ + │ POST /device/code │ │ + │ client_id=lwlpk_U7Qy7ThG69STZk │ │ + │ scope=userinfo:read │ │ + │ payment_methods.agentic │ │ + │─────────────────────────────────────>│ │ + │ │ │ + │ { device_code, user_code, │ │ + │ verification_uri, │ │ + │ verification_uri_complete } │ │ + │<─────────────────────────────────────│ │ + │ │ │ + │ Show verification_uri to user ──────┼──────────────────────────────────>│ + │ │ │ + │ │ User visits URL, logs in, │ + │ │ enters user_code phrase │ + │ │<─────────────────────────────────│ + │ │ │ + │ POST /device/token (poll) │ │ + │ grant_type=device_code │ │ + │ device_code=... │ │ + │─────────────────────────────────────>│ │ + │ │ │ + │ { access_token, refresh_token, │ │ + │ expires_in, token_type } │ │ + │<─────────────────────────────────────│ │ + │ │ │ +``` + +**Key details:** +- **Client ID**: Hardcoded public `lwlpk_U7Qy7ThG69STZk` (no client_secret needed) +- **Scopes**: `userinfo:read payment_methods.agentic` +- **Token refresh**: Standard `refresh_token` grant at same `/device/token` endpoint +- **API calls**: Bearer token in `Authorization` header to `api.link.com` +- **User approval**: Push notification or email via Link app, with code-phrase confirmation + +## AutoGPT's Current OAuth Model + +AutoGPT uses the **Authorization Code Grant** flow: + +1. Backend generates `login_url` → frontend redirects user to provider +2. User authorizes → provider redirects back with `code` +3. Backend exchanges `code` for tokens via `POST /{provider}/callback` +4. Tokens stored as `OAuth2Credentials` (access_token + refresh_token) + +The `BaseOAuthHandler` interface: +```python +class BaseOAuthHandler(ABC): + def __init__(self, client_id, client_secret, redirect_uri): ... + def get_login_url(self, scopes, state, code_challenge) -> str: ... + async def exchange_code_for_tokens(self, code, scopes, code_verifier) -> OAuth2Credentials: ... + async def _refresh_tokens(self, credentials) -> OAuth2Credentials: ... + async def revoke_tokens(self, credentials) -> bool: ... +``` + +## The Mismatch + +| Aspect | Authorization Code (current) | Device Code (Link CLI) | +|--------|------------------------------|----------------------| +| **Initiation** | Redirect user to login URL | Show verification URL + code phrase | +| **Token acquisition** | One-shot callback with `code` | Poll `/device/token` until approved | +| **Client secret** | Required | Not used (public client) | +| **Redirect URI** | Required | Not used | +| **User interaction** | Same browser (redirect) | Separate device (phone/other browser) | + +## Implementation Options + +### Option A: Adapt `BaseOAuthHandler` (Minimal Backend Changes) + +Map the device code flow onto the existing handler interface: + +- `get_login_url()` → call `/device/code`, return `verification_uri_complete` + - Store `device_code` in the OAuth state token +- `exchange_code_for_tokens()` → poll `/device/token` with the stored `device_code` + - The `code` parameter is repurposed as the device_code +- `_refresh_tokens()` → standard refresh_token grant ✅ +- `revoke_tokens()` → call `/device/revoke` ✅ + +**Pros:** Minimal backend changes, reuses existing OAuth infrastructure +**Cons:** +- Frontend redirect UX doesn't match — need to show "visit URL" instead of redirect +- Polling doesn't fit the one-shot callback model — `exchange_code_for_tokens` would + need to block/poll (potentially for minutes) +- The frontend currently opens a popup/redirect; it would need a new "device auth" UI mode + +### Option B: API Key Credential (Simplest) + +Let users paste a pre-obtained access token as an API key: + +```python +StripeLinkCredentials = APIKeyCredentials +StripeLinkCredentialsInput = CredentialsMetaInput[ + Literal[ProviderName.STRIPE_LINK], Literal["api_key"] +] +``` + +**Pros:** Zero infrastructure changes, works today +**Cons:** +- Terrible UX — user must run `link-cli auth login` externally, copy token +- No auto-refresh — tokens expire, user must re-authenticate manually +- Defeats the purpose of integrated credential management + +### Option C: New Device Code OAuth Flow (Recommended) + +Add a **device code flow variant** to the integrations system: + +1. **New backend endpoint**: `POST /integrations/{provider}/device-auth` + - Calls Link's `/device/code` + - Returns `{ verification_url, user_code, poll_token }` to frontend +2. **New backend endpoint**: `GET /integrations/{provider}/device-auth/poll` + - Polls Link's `/device/token` on demand + - Returns `{ status: "pending" | "approved" | "denied" }` + - On "approved", stores `OAuth2Credentials` and returns credential metadata +3. **Frontend UI**: Show verification URL + code phrase, poll status +4. **OAuth handler**: New `BaseDeviceAuthHandler` base class + +```python +class BaseDeviceAuthHandler(ABC): + """Handler for OAuth 2.0 Device Code Grant flows""" + PROVIDER_NAME: ClassVar[ProviderName | str] + + @abstractmethod + async def initiate_device_auth(self, scopes: list[str]) -> DeviceAuthState: ... + + @abstractmethod + async def poll_device_auth(self, device_code: str) -> OAuth2Credentials | None: ... + + @abstractmethod + async def _refresh_tokens(self, credentials: OAuth2Credentials) -> OAuth2Credentials: ... + + @abstractmethod + async def revoke_tokens(self, credentials: OAuth2Credentials) -> bool: ... +``` + +**Pros:** +- Clean separation of concerns +- Reusable for other device-code providers (smart TVs, CLI tools, IoT) +- Good UX — user sees clear instructions, approval via Link app +- Auto-refresh works via standard OAuth2Credentials + +**Cons:** +- Requires new API endpoints and frontend UI components +- More implementation effort upfront + +### Option D: Backend Polling with SSE/WebSocket (Best UX, Most Complex) + +Like Option C but the backend polls automatically and pushes status to frontend via SSE: + +- Backend initiates device auth and starts polling in a background task +- Frontend connects via SSE or WebSocket for real-time status updates +- When approved, credentials are stored and frontend is notified instantly + +**Pros:** Best UX (no frontend polling), instant notification +**Cons:** Significant complexity, SSE/WebSocket infrastructure needed + +## Recommendation + +**Start with Option C** (New Device Code OAuth Flow). It's the cleanest architecture +that properly handles the device code flow without hacking it into the authorization +code flow. The endpoints and handler abstraction are also reusable for future providers. + +**Fallback to Option A** if we want a quick proof of concept — the main challenge is +frontend UX, but the backend adaptation is relatively straightforward. + +## What Blocks Would Look Like + +Regardless of auth approach, the block surface area would be: + +| Block | Description | Auth Scope | +|-------|-------------|------------| +| `StripeLinkListPaymentMethodsBlock` | List cards/bank accounts | `payment_methods.agentic` | +| `StripeLinkCreateSpendRequestBlock` | Create a spend request | `payment_methods.agentic` | +| `StripeLinkRetrieveSpendRequestBlock` | Get spend request + card details | `payment_methods.agentic` | +| `StripeLinkRequestApprovalBlock` | Request user approval for spend | `payment_methods.agentic` | +| `StripeLinkMPPPayBlock` | Execute MPP payment | `payment_methods.agentic` | + +All blocks would use the same credential type: +```python +StripeLinkCredentials = OAuth2Credentials +StripeLinkCredentialsInput = CredentialsMetaInput[ + Literal[ProviderName.STRIPE_LINK], Literal["oauth2"] +] +``` + +See `_auth.py` and `spend_request.py` in this directory for skeleton implementations. diff --git a/autogpt_platform/backend/backend/blocks/stripe_link/__init__.py b/autogpt_platform/backend/backend/blocks/stripe_link/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/autogpt_platform/backend/backend/blocks/stripe_link/_auth.py b/autogpt_platform/backend/backend/blocks/stripe_link/_auth.py new file mode 100644 index 0000000000..a574b2dc31 --- /dev/null +++ b/autogpt_platform/backend/backend/blocks/stripe_link/_auth.py @@ -0,0 +1,99 @@ +""" +Stripe Link CLI — Credential definitions for AutoGPT blocks. + +This file defines the credential types and helpers that Stripe Link blocks use. +Link CLI uses OAuth 2.0 Device Code Grant (RFC 8628), which produces standard +access_token + refresh_token pairs — so we store them as OAuth2Credentials. + +AUTH FLOW NOTES: + Link's auth is a Device Code flow (not Authorization Code): + 1. POST https://login.link.com/device/code + → { device_code, user_code, verification_uri } + 2. User visits verification_uri, logs in, enters code phrase + 3. Poll POST https://login.link.com/device/token + → { access_token, refresh_token, expires_in } + + This doesn't fit the current BaseOAuthHandler interface (which assumes + Authorization Code Grant with redirects). See EXPLORATION.md for options. + The recommended path is adding a BaseDeviceAuthHandler abstraction. + + For now, this file defines the credential *shape* that blocks expect, + independent of how the credentials are obtained. +""" + +from typing import Literal + +from pydantic import SecretStr + +from backend.data.model import ( + CredentialsField, + CredentialsMetaInput, + OAuth2Credentials, +) +from backend.integrations.providers import ProviderName + +# --------------------------------------------------------------------------- +# Provider name — would need to be added to the ProviderName enum, but the +# enum supports dynamic values via _missing_() so this works without changes. +# --------------------------------------------------------------------------- +STRIPE_LINK_PROVIDER = ProviderName("stripe_link") + +# --------------------------------------------------------------------------- +# Link API constants +# --------------------------------------------------------------------------- +LINK_AUTH_BASE_URL = "https://login.link.com" +LINK_API_BASE_URL = "https://api.link.com" +LINK_CLIENT_ID = "lwlpk_U7Qy7ThG69STZk" +LINK_DEFAULT_SCOPES = ["userinfo:read", "payment_methods.agentic"] + +# --------------------------------------------------------------------------- +# Credential type definitions +# +# Link CLI produces OAuth2 tokens (access_token + refresh_token), so we use +# OAuth2Credentials. The blocks don't care how the tokens were obtained — +# they just need a valid access_token for Bearer auth against api.link.com. +# --------------------------------------------------------------------------- +StripeLinkCredentials = OAuth2Credentials + +StripeLinkCredentialsInput = CredentialsMetaInput[ + Literal[ProviderName.STRIPE_LINK], # type: ignore[index] + Literal["oauth2"], +] + + +def StripeLinkCredentialsField() -> StripeLinkCredentialsInput: + """ + Creates a Stripe Link credentials input on a block. + + All Link blocks require the same `payment_methods.agentic` scope. + """ + return CredentialsField( + required_scopes=set(LINK_DEFAULT_SCOPES), + description=( + "Connect your Stripe Link account to enable the agent to request " + "secure, one-time-use payment credentials from your Link wallet. " + "You'll approve each spend request via the Link app." + ), + ) + + +# --------------------------------------------------------------------------- +# Test credentials for block testing +# --------------------------------------------------------------------------- +TEST_CREDENTIALS = OAuth2Credentials( + id="01234567-89ab-cdef-0123-456789abcdef", + provider="stripe_link", + access_token=SecretStr("mock-link-access-token"), + refresh_token=SecretStr("mock-link-refresh-token"), + access_token_expires_at=None, # Won't expire in tests + scopes=LINK_DEFAULT_SCOPES, + title="Mock Stripe Link credentials", + username="test@example.com", +) + +TEST_CREDENTIALS_INPUT = { + "provider": TEST_CREDENTIALS.provider, + "id": TEST_CREDENTIALS.id, + "type": TEST_CREDENTIALS.type, + "title": TEST_CREDENTIALS.title, +} diff --git a/autogpt_platform/backend/backend/blocks/stripe_link/_device_auth_handler.py b/autogpt_platform/backend/backend/blocks/stripe_link/_device_auth_handler.py new file mode 100644 index 0000000000..7a7c3f3d77 --- /dev/null +++ b/autogpt_platform/backend/backend/blocks/stripe_link/_device_auth_handler.py @@ -0,0 +1,206 @@ +""" +Skeleton for a Device Code OAuth handler for Stripe Link. + +This would live in backend/integrations/oauth/ once the device-code flow +is officially supported. It shows what an implementation of Option C from +EXPLORATION.md would look like. + +NOTE: This is exploration code — it won't work until: + 1. BaseDeviceAuthHandler (or equivalent) is added to the framework + 2. New API endpoints for device-auth initiation and polling are added + 3. Frontend UI for the device-code flow is implemented +""" + +import logging +import time +from dataclasses import dataclass +from typing import ClassVar + +import httpx +from pydantic import SecretStr + +from backend.data.model import OAuth2Credentials +from backend.integrations.providers import ProviderName + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- +LINK_AUTH_BASE_URL = "https://login.link.com" +LINK_CLIENT_ID = "lwlpk_U7Qy7ThG69STZk" +DEFAULT_SCOPES = ["userinfo:read", "payment_methods.agentic"] + + +# --------------------------------------------------------------------------- +# Data classes for the device code flow +# --------------------------------------------------------------------------- +@dataclass +class DeviceAuthInitiation: + """Returned when initiating the device code flow.""" + + device_code: str + user_code: str + verification_url: str + verification_url_complete: str + expires_in: int + interval: int # seconds between polls + + +# --------------------------------------------------------------------------- +# Handler implementation +# +# This CANNOT extend BaseOAuthHandler because the interface doesn't match. +# Instead, this shows what a new BaseDeviceAuthHandler or a standalone +# handler class would look like. +# --------------------------------------------------------------------------- +class StripeLinkDeviceAuthHandler: + """ + Handles the OAuth 2.0 Device Code Grant for Stripe Link. + + Flow: + 1. initiate_device_auth() → DeviceAuthInitiation + - POST /device/code with client_id + scope + - Returns verification URL for user + device_code for polling + + 2. poll_for_tokens(device_code) → OAuth2Credentials | None + - POST /device/token with grant_type=device_code + - Returns None while pending, raises on denial/expiry + - Returns OAuth2Credentials on approval + + 3. refresh_tokens(credentials) → OAuth2Credentials + - POST /device/token with grant_type=refresh_token + - Standard refresh flow + + 4. revoke_tokens(credentials) → bool + - POST /device/revoke + """ + + PROVIDER_NAME: ClassVar[ProviderName] = ProviderName("stripe_link") + + async def initiate_device_auth( + self, + scopes: list[str] | None = None, + client_name: str = "AutoGPT", + ) -> DeviceAuthInitiation: + """Start the device code flow. Returns URLs for the user to visit.""" + import socket + + effective_scopes = " ".join(scopes or DEFAULT_SCOPES) + hostname = socket.gethostname() + + async with httpx.AsyncClient() as client: + response = await client.post( + f"{LINK_AUTH_BASE_URL}/device/code", + data={ + "client_id": LINK_CLIENT_ID, + "scope": effective_scopes, + "connection_label": f"{client_name} on {hostname}", + "client_hint": client_name, + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + response.raise_for_status() + data = response.json() + + return DeviceAuthInitiation( + device_code=data["device_code"], + user_code=data["user_code"], + verification_url=data["verification_uri"], + verification_url_complete=data["verification_uri_complete"], + expires_in=data["expires_in"], + interval=data["interval"], + ) + + async def poll_for_tokens(self, device_code: str) -> OAuth2Credentials | None: + """ + Poll for token completion. Returns None if still pending. + Raises on expiry or denial. + """ + async with httpx.AsyncClient() as client: + response = await client.post( + f"{LINK_AUTH_BASE_URL}/device/token", + data={ + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + "device_code": device_code, + "client_id": LINK_CLIENT_ID, + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + if response.status_code == 200: + data = response.json() + return OAuth2Credentials( + provider=self.PROVIDER_NAME, + access_token=SecretStr(data["access_token"]), + refresh_token=SecretStr(data["refresh_token"]), + access_token_expires_at=int(time.time()) + data["expires_in"], + scopes=DEFAULT_SCOPES, + title="Stripe Link", + ) + + if response.status_code == 400: + error = response.json() + error_code = error.get("error", "") + + if error_code in ("authorization_pending", "slow_down"): + return None # Still waiting for user + + if error_code == "expired_token": + raise RuntimeError( + "Device code expired. Please restart the login flow." + ) + + if error_code == "access_denied": + raise RuntimeError("Authorization denied by user.") + + raise RuntimeError( + f"Unexpected response from Link auth: {response.status_code} " + f"{response.text}" + ) + + async def refresh_tokens(self, credentials: OAuth2Credentials) -> OAuth2Credentials: + """Refresh an expired access token using the refresh token.""" + if not credentials.refresh_token: + raise RuntimeError("No refresh token available") + + async with httpx.AsyncClient() as client: + response = await client.post( + f"{LINK_AUTH_BASE_URL}/device/token", + data={ + "grant_type": "refresh_token", + "refresh_token": credentials.refresh_token.get_secret_value(), + "client_id": LINK_CLIENT_ID, + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + response.raise_for_status() + data = response.json() + + credentials.access_token = SecretStr(data["access_token"]) + credentials.refresh_token = SecretStr(data["refresh_token"]) + credentials.access_token_expires_at = int(time.time()) + data["expires_in"] + return credentials + + async def revoke_tokens(self, credentials: OAuth2Credentials) -> bool: + """Revoke the refresh token at the Link auth server.""" + if not credentials.refresh_token: + return False + + async with httpx.AsyncClient() as client: + response = await client.post( + f"{LINK_AUTH_BASE_URL}/device/revoke", + data={ + "client_id": LINK_CLIENT_ID, + "token": credentials.refresh_token.get_secret_value(), + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + + return response.status_code == 200 + + def needs_refresh(self, credentials: OAuth2Credentials) -> bool: + """Check if the access token needs refreshing (5-minute window).""" + if credentials.access_token_expires_at is None: + return False + return credentials.access_token_expires_at < int(time.time()) + 300 diff --git a/autogpt_platform/backend/backend/blocks/stripe_link/spend_request.py b/autogpt_platform/backend/backend/blocks/stripe_link/spend_request.py new file mode 100644 index 0000000000..be54c908b9 --- /dev/null +++ b/autogpt_platform/backend/backend/blocks/stripe_link/spend_request.py @@ -0,0 +1,375 @@ +""" +Stripe Link — Spend Request blocks. + +These blocks interact with the Link API (api.link.com) to create, retrieve, +and approve spend requests. A spend request provisions a one-time-use virtual +card or shared payment token from the user's Link wallet. + +IMPORTANT: These are SKELETON implementations for exploration. The HTTP client +logic (calling api.link.com) needs to be fleshed out, and the device-code auth +flow needs to land before these can actually run. +""" + +import logging +from typing import Any + + +from backend.blocks.stripe_link._auth import ( + TEST_CREDENTIALS, + TEST_CREDENTIALS_INPUT, + StripeLinkCredentials, + StripeLinkCredentialsField, + StripeLinkCredentialsInput, +) +from backend.data.block import Block, BlockOutput, BlockSchemaInput +from backend.data.model import SchemaField + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Shared helpers +# --------------------------------------------------------------------------- +LINK_API_BASE = "https://api.link.com" + + +async def _link_api_request( + credentials: StripeLinkCredentials, + method: str, + path: str, + body: dict[str, Any] | None = None, +) -> dict[str, Any]: + """ + Make an authenticated request to the Link API. + + Uses the access_token from OAuth2Credentials as a Bearer token. + In a real implementation, this should handle 401 → token refresh. + """ + import httpx + + headers = { + "Authorization": f"Bearer {credentials.access_token.get_secret_value()}", + "Content-Type": "application/json", + } + + async with httpx.AsyncClient() as client: + response = await client.request( + method=method, + url=f"{LINK_API_BASE}{path}", + headers=headers, + json=body, + ) + response.raise_for_status() + return response.json() + + +# --------------------------------------------------------------------------- +# Block: List Payment Methods +# --------------------------------------------------------------------------- +class StripeLinkListPaymentMethodsBlock(Block): + """List payment methods (cards and bank accounts) from the user's Link wallet.""" + + class Input(BlockSchemaInput): + credentials: StripeLinkCredentialsInput = StripeLinkCredentialsField() + + class Output(BlockSchemaInput): + payment_methods: list[dict[str, Any]] = SchemaField( + description="List of payment methods in the Link wallet" + ) + error: str = SchemaField( + description="Error message if the request failed", + default="", + ) + + def __init__(self): + super().__init__( + id="a1b2c3d4-e5f6-7890-abcd-ef1234567890", # placeholder UUID + description="List payment methods from a Stripe Link wallet", + categories=set(), + input_schema=self.Input, + output_schema=self.Output, + test_input={ + "credentials": TEST_CREDENTIALS_INPUT, + }, + test_credentials=TEST_CREDENTIALS, + test_output=[ + ( + "payment_methods", + [ + { + "id": "csmrpd_test", + "type": "card", + "is_default": True, + "card_details": { + "brand": "visa", + "last4": "4242", + "exp_month": 12, + "exp_year": 2030, + }, + } + ], + ) + ], + test_mock={ + "_link_api_request": lambda *args, **kwargs: [ + { + "id": "csmrpd_test", + "type": "card", + "is_default": True, + "card_details": { + "brand": "visa", + "last4": "4242", + "exp_month": 12, + "exp_year": 2030, + }, + } + ] + }, + ) + + async def run( + self, + input_data: Input, + *, + credentials: StripeLinkCredentials, + **kwargs: Any, + ) -> BlockOutput: + try: + methods = await _link_api_request(credentials, "GET", "/payment_methods") + yield "payment_methods", methods + except Exception as e: + yield "error", str(e) + + +# --------------------------------------------------------------------------- +# Block: Create Spend Request +# --------------------------------------------------------------------------- +class StripeLinkCreateSpendRequestBlock(Block): + """ + Create a spend request to get a one-time-use payment credential. + + The user must approve the request via the Link app before card details + are available. Use StripeLinkRetrieveSpendRequestBlock to check status + and get the credential once approved. + """ + + class Input(BlockSchemaInput): + credentials: StripeLinkCredentialsInput = StripeLinkCredentialsField() + payment_method_id: str = SchemaField( + description="ID of the payment method to use (from list payment methods)" + ) + merchant_name: str = SchemaField( + description="Name of the merchant for this purchase" + ) + merchant_url: str = SchemaField(description="URL of the merchant website") + context: str = SchemaField( + description=( + "Description of the purchase context (min 100 characters). " + "Shown to the user when they approve the request." + ) + ) + amount: int = SchemaField( + description="Amount in cents (max 50000)", ge=1, le=50000 + ) + currency: str = SchemaField( + description="3-letter ISO currency code", default="usd" + ) + request_approval: bool = SchemaField( + description=( + "If true, immediately sends a push notification to the user " + "for approval. Otherwise, call request-approval separately." + ), + default=True, + ) + test_mode: bool = SchemaField( + description="Use test mode (fake card 4242424242424242)", + default=False, + ) + + class Output(BlockSchemaInput): + spend_request_id: str = SchemaField( + description="ID of the created spend request" + ) + status: str = SchemaField( + description="Status: created, pending_approval, approved, denied, etc." + ) + approval_url: str = SchemaField( + description="URL the user can visit to approve (if not using push)", + default="", + ) + error: str = SchemaField( + description="Error message if the request failed", + default="", + ) + + def __init__(self): + super().__init__( + id="b2c3d4e5-f6a7-8901-bcde-f12345678901", # placeholder UUID + description="Create a Stripe Link spend request for a one-time payment credential", + categories=set(), + input_schema=self.Input, + output_schema=self.Output, + test_input={ + "credentials": TEST_CREDENTIALS_INPUT, + "payment_method_id": "csmrpd_test", + "merchant_name": "Test Store", + "merchant_url": "https://example.com", + "context": "x" * 100, + "amount": 1000, + }, + test_credentials=TEST_CREDENTIALS, + test_output=[ + ("spend_request_id", "lsrq_test123"), + ("status", "pending_approval"), + ], + test_mock={ + "_link_api_request": lambda *args, **kwargs: { + "id": "lsrq_test123", + "status": "pending_approval", + "approval_url": "", + } + }, + ) + + async def run( + self, + input_data: Input, + *, + credentials: StripeLinkCredentials, + **kwargs: Any, + ) -> BlockOutput: + try: + result = await _link_api_request( + credentials, + "POST", + "/spend_requests", + body={ + "payment_details": input_data.payment_method_id, + "merchant_name": input_data.merchant_name, + "merchant_url": input_data.merchant_url, + "context": input_data.context, + "amount": input_data.amount, + "currency": input_data.currency, + "request_approval": input_data.request_approval, + "test": input_data.test_mode, + }, + ) + yield "spend_request_id", result["id"] + yield "status", result["status"] + if result.get("approval_url"): + yield "approval_url", result["approval_url"] + except Exception as e: + yield "error", str(e) + + +# --------------------------------------------------------------------------- +# Block: Retrieve Spend Request +# --------------------------------------------------------------------------- +class StripeLinkRetrieveSpendRequestBlock(Block): + """ + Retrieve a spend request and its credentials (once approved). + + After the user approves a spend request, this block returns the + virtual card details (number, CVC, expiry, billing address) that + can be used for a one-time purchase. + """ + + class Input(BlockSchemaInput): + credentials: StripeLinkCredentialsInput = StripeLinkCredentialsField() + spend_request_id: str = SchemaField( + description="ID of the spend request to retrieve (e.g., lsrq_...)" + ) + include_card: bool = SchemaField( + description="Include unmasked card details in the response", + default=True, + ) + + class Output(BlockSchemaInput): + status: str = SchemaField(description="Current status of the spend request") + card_number: str = SchemaField( + description="Virtual card number (only if approved and include_card=True)", + default="", + ) + card_cvc: str = SchemaField( + description="Virtual card CVC", + default="", + ) + card_exp_month: int = SchemaField( + description="Card expiry month", + default=0, + ) + card_exp_year: int = SchemaField( + description="Card expiry year", + default=0, + ) + card_brand: str = SchemaField( + description="Card brand (visa, mastercard, etc.)", + default="", + ) + valid_until: str = SchemaField( + description="ISO timestamp when the virtual card expires", + default="", + ) + error: str = SchemaField( + description="Error message if the request failed", + default="", + ) + + def __init__(self): + super().__init__( + id="c3d4e5f6-a7b8-9012-cdef-123456789012", # placeholder UUID + description="Retrieve a Stripe Link spend request and card credentials", + categories=set(), + input_schema=self.Input, + output_schema=self.Output, + test_input={ + "credentials": TEST_CREDENTIALS_INPUT, + "spend_request_id": "lsrq_test123", + }, + test_credentials=TEST_CREDENTIALS, + test_output=[ + ("status", "approved"), + ("card_number", "4242424242424242"), + ], + test_mock={ + "_link_api_request": lambda *args, **kwargs: { + "status": "approved", + "card": { + "number": "4242424242424242", + "cvc": "123", + "exp_month": 12, + "exp_year": 2030, + "brand": "visa", + "valid_until": "2025-12-31T23:59:59Z", + }, + } + }, + ) + + async def run( + self, + input_data: Input, + *, + credentials: StripeLinkCredentials, + **kwargs: Any, + ) -> BlockOutput: + try: + include = ["card"] if input_data.include_card else [] + path = f"/spend_requests/{input_data.spend_request_id}" + if include: + path += f"?include={','.join(include)}" + + result = await _link_api_request(credentials, "GET", path) + + yield "status", result["status"] + + card = result.get("card") + if card: + yield "card_number", card.get("number", "") + yield "card_cvc", card.get("cvc", "") + yield "card_exp_month", card.get("exp_month", 0) + yield "card_exp_year", card.get("exp_year", 0) + yield "card_brand", card.get("brand", "") + yield "valid_until", card.get("valid_until", "") + except Exception as e: + yield "error", str(e)