github/AutoGPT
1
1
Fork 0
mirror of https://github.com/Significant-Gravitas/AutoGPT.git synced 2026-04-30 03:00:41 -04:00
Code Issues Packages Projects Releases Wiki Activity

explore(blocks): Stripe Link CLI block auth exploration

Browse Source 
Add skeleton code exploring what a Stripe Link CLI integration would
look like from an auth perspective. Link CLI uses OAuth 2.0 Device Code
Grant (RFC 8628), which doesn't map directly onto AutoGPT's current
Authorization Code Grant-based OAuth handler system.

Files:
- EXPLORATION.md: Detailed analysis of 4 auth integration options
- _auth.py: Credential type definitions (OAuth2Credentials)
- _device_auth_handler.py: Skeleton Device Code flow handler
- spend_request.py: Skeleton blocks (list methods, create/retrieve spend)

Key finding: Recommend adding a BaseDeviceAuthHandler abstraction (Option C)
to properly support the device code flow, which is also reusable for other
CLI/IoT-style OAuth providers.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Nicholas Tindle
2026-04-29 17:12:14 -05:00
parent c08b9774dc
commit 24f7a9d28e
5 changed files with 890 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

210
autogpt_platform/backend/backend/blocks/stripe_link/EXPLORATION.md Normal file
View File

@@ -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.

0
autogpt_platform/backend/backend/blocks/stripe_link/__init__.py Normal file
View File

99
autogpt_platform/backend/backend/blocks/stripe_link/_auth.py Normal file
View File

@@ -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,
}

206
autogpt_platform/backend/backend/blocks/stripe_link/_device_auth_handler.py Normal file
View File

@@ -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

375
autogpt_platform/backend/backend/blocks/stripe_link/spend_request.py Normal file
View File

@@ -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)