/**
* Python SDK Generator
* Generates Python client SDKs for x402-enabled APIs
*
* @author nich
* @github github.com/nirholas
* @license Apache-2.0
*/
import {
SDKConfig,
SDKGenerationResult,
toPascalCase,
toCamelCase,
toSnakeCase,
extractPathParams,
} from './types.js';
/**
* Generate Python method name from HTTP method and path
*/
function routeToSnakeCaseMethod(method: string, path: string): string {
const parts = path.split('/').filter(p => p && !p.startsWith(':'));
const name = parts.map(p => toSnakeCase(p)).join('_');
return method.toLowerCase() + (name ? '_' + name : '');
}
/**
* Generate Python route methods
*/
function generateRouteMethodsPython(routes: Record<string, string>): string {
const methods: string[] = [];
for (const [route, price] of Object.entries(routes)) {
const [method, path] = route.split(' ');
const methodName = routeToSnakeCaseMethod(method, path);
const params = extractPathParams(path);
const paramList = params.map(p => `${p}: str`).join(', ');
const bodyParam = ['POST', 'PUT', 'PATCH'].includes(method)
? (paramList ? ', body: Optional[Dict[str, Any]] = None' : 'body: Optional[Dict[str, Any]] = None')
: '';
const pathWithParams = params.reduce(
(p, param) => p.replace(`:${param}`, `{${param}}`),
path
);
methods.push(` def ${methodName}(self${paramList ? ', ' + paramList : ''}${bodyParam}) -> Any:
"""
${method} ${path}
Price: $${price}
${params.length > 0 ? 'Args:\n' + params.map(p => ` ${p}: Path parameter`).join('\n') + '\n ' : ''}${bodyParam ? 'Args:\n body: Request body (optional)\n ' : ''}
Returns:
API response data
"""
return self._make_request(
"${method}",
f"${pathWithParams}",
"${price}"${bodyParam ? ', body' : ''}
)`);
}
return methods.join('\n\n');
}
/**
* Generate a Python SDK for the given configuration
*/
export function generatePythonSDK(config: SDKConfig): SDKGenerationResult {
const clientName = toPascalCase(config.apiName);
const instanceName = toCamelCase(config.apiName);
const code = `"""
${config.apiName} Python SDK
Auto-generated by x402-deploy
@see https://x402.org
"""
import os
import requests
from typing import Optional, Dict, Any, TypedDict
from datetime import datetime
class PaymentProof(TypedDict):
"""Payment proof structure"""
signature: str
timestamp: int
amount: str
network: str
class APIError(Exception):
"""API error with status code and message"""
def __init__(self, status_code: int, message: str):
self.status_code = status_code
self.message = message
super().__init__(f"API error {status_code}: {message}")
class ${clientName}Client:
"""
${config.apiName} API Client
This client handles x402 payment authentication automatically.
Example:
client = ${clientName}Client(payer_private_key=os.environ['PAYER_KEY'])
result = client.get_data('123')
"""
def __init__(
self,
facilitator_url: str = "${config.facilitator}",
payer_private_key: Optional[str] = None,
timeout: int = 30
):
"""
Initialize the ${config.apiName} client.
Args:
facilitator_url: URL of the x402 facilitator service
payer_private_key: Private key for automatic payments (optional)
timeout: Request timeout in seconds
"""
self.api_url = "${config.apiUrl}"
self.facilitator_url = facilitator_url
self.payer_private_key = payer_private_key
self.wallet = "${config.wallet}"
self.timeout = timeout
self._session = requests.Session()
def _get_payment_proof(self, route: str, price: str) -> str:
"""
Get payment proof from facilitator.
Args:
route: The API route being accessed
price: The price in USD
Returns:
Payment proof string
"""
payload = {
"recipient": self.wallet,
"amount": price,
"network": "eip155:8453",
"route": route
}
if self.payer_private_key:
payload["payerKey"] = self.payer_private_key
response = self._session.post(
f"{self.facilitator_url}/create-payment",
json=payload,
timeout=self.timeout
)
response.raise_for_status()
return response.json()["proof"]
def _make_request(
self,
method: str,
path: str,
price: str,
body: Optional[Dict[str, Any]] = None
) -> Any:
"""
Make authenticated request to API.
Args:
method: HTTP method
path: API path
price: Price in USD
body: Request body (optional)
Returns:
API response data
"""
proof = self._get_payment_proof(f"{method} {path}", price)
response = self._session.request(
method=method,
url=f"{self.api_url}{path}",
headers={
"Content-Type": "application/json",
"x-payment": proof
},
json=body,
timeout=self.timeout
)
if not response.ok:
raise APIError(response.status_code, response.text)
return response.json()
def get_discovery(self) -> Dict[str, Any]:
"""
Get the x402 discovery document.
Returns:
Discovery document with pricing and configuration
"""
response = self._session.get(
f"{self.api_url}/.well-known/x402",
timeout=self.timeout
)
response.raise_for_status()
return response.json()
def close(self) -> None:
"""Close the HTTP session."""
self._session.close()
def __enter__(self) -> "${clientName}Client":
return self
def __exit__(self, *args) -> None:
self.close()
${generateRouteMethodsPython(config.routes)}
# Create default client instance
${instanceName} = ${clientName}Client()
# Convenience function for one-off requests
def create_client(
facilitator_url: str = "${config.facilitator}",
payer_private_key: Optional[str] = None
) -> ${clientName}Client:
"""
Create a new ${config.apiName} client.
Args:
facilitator_url: URL of the x402 facilitator service
payer_private_key: Private key for automatic payments
Returns:
Configured client instance
"""
return ${clientName}Client(
facilitator_url=facilitator_url,
payer_private_key=payer_private_key
)
`;
return {
language: 'python',
code,
extension: 'py',
directory: 'python'
};
}
