File size: 8,985 Bytes
c5f454e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 |
# Simple client to interact with AgentHub API
import requests
from typing import Optional, Dict, Any
import json # For decoding JSON response
from .types import ConsumptionResult
from .exceptions import AgentPayAPIError, AgentPayConnectionError # Import custom exceptions
API_BASE_URL = "https://api.agentpay.me"
API_VERSION_PREFIX = "/v1"
class AgentPayClient:
"""Client for interacting with the AgentPay API.
Handles authentication using a Service Token and provides methods
for validating user API keys and consuming balance.
"""
def __init__(
self,
service_token: str,
api_url: Optional[str] = None,
):
"""Initializes the AgentPay Client.
Args:
service_token: The unique token issued to your service by AgentHub.
api_url: Optional. Manually override the base API URL.
"""
if not service_token:
raise ValueError("service_token is required.")
self.service_token = service_token
if api_url:
# Use manual override if provided
self.base_api_url = api_url.rstrip('/')
else:
self.base_api_url = API_BASE_URL
# Add the /api/v1 prefix
self.base_api_url += API_VERSION_PREFIX
# Prepare headers for requests
self.headers = {
"Authorization": f"Bearer {self.service_token}",
"Content-Type": "application/json",
"Accept": "application/json",
}
def _request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Internal helper to make requests to the AgentPay API.
Args:
method: HTTP method (e.g., "POST", "GET").
endpoint: API endpoint path (e.g., "/balances/consume").
data: Optional dictionary payload for POST/PUT/PATCH requests.
Returns:
The JSON response body as a dictionary.
Raises:
AgentPayConnectionError: If a network error occurs.
AgentPayAPIError: If the API returns an error status code.
"""
full_url = self.base_api_url + endpoint
try:
response = requests.request(
method=method,
url=full_url,
headers=self.headers,
json=data, # requests library handles JSON serialization
timeout=10 # Add a reasonable timeout (e.g., 10 seconds)
)
# Raise exception for bad status codes (4xx or 5xx)
response.raise_for_status()
# Attempt to parse successful response as JSON
# Handle cases where response might be empty (e.g., 204 No Content)
if response.status_code == 204:
return {} # Return empty dict for No Content
return response.json()
except requests.exceptions.Timeout as e:
raise AgentPayConnectionError(f"Request timed out connecting to {full_url}: {e}") from e
except requests.exceptions.ConnectionError as e:
raise AgentPayConnectionError(f"Could not connect to {full_url}: {e}") from e
except requests.exceptions.HTTPError as e:
# Handle API errors (4xx, 5xx)
status_code = e.response.status_code
error_code = None
error_message = None
try:
# Attempt to parse error details from response body
error_data = e.response.json()
error_code = error_data.get("error_code")
error_message = error_data.get("error_message") or error_data.get("detail") # FastAPI uses detail sometimes
except json.JSONDecodeError:
# If response body is not JSON or empty
error_message = e.response.text or f"HTTP error {status_code}"
raise AgentPayAPIError(
status_code=status_code,
error_code=error_code,
error_message=error_message
) from e
except json.JSONDecodeError as e:
# Handle cases where a 2xx response is not valid JSON
raise AgentPayAPIError(
status_code=response.status_code, # Use the status code from the original response
error_message=f"Failed to decode successful JSON response: {e}"
) from e
except requests.exceptions.RequestException as e:
# Catch any other requests-related errors
raise AgentPayConnectionError(f"An unexpected network error occurred: {e}") from e
def consume(
self,
api_key: str,
amount_cents: int,
usage_event_id: str,
metadata: Optional[Dict[str, Any]] = None,
) -> ConsumptionResult:
"""Consumes balance for a user associated with an API key.
Args:
api_key: The User's API key to consume balance against.
amount_cents: The amount to consume in cents (must be >= 0).
usage_event_id: A unique identifier (e.g., UUID string) for this specific
consumption attempt, used for idempotency.
metadata: Optional dictionary containing additional data to associate
with the usage event.
Returns:
A ConsumptionResult object indicating success or failure.
"""
if amount_cents < 0:
# Prevent making an API call with an invalid amount
# The API would reject this anyway, but better to catch early.
return ConsumptionResult(
success=False,
error_code="INVALID_INPUT",
error_message="Consumption amount cannot be negative."
)
endpoint = "/balances/consume"
payload = {
"api_key": api_key,
"amount_cents": amount_cents,
"usage_event_id": usage_event_id,
}
if metadata:
payload["metadata"] = metadata
try:
# _request raises AgentPayAPIError for 4xx/5xx responses
# and AgentPayConnectionError for network issues.
response_data = self._request("POST", endpoint, data=payload)
# If _request completes without raising an exception, it implies a 2xx response.
# The API's 200 OK response for consume just confirms success.
return ConsumptionResult(success=True)
except AgentPayAPIError as e:
# API returned a non-2xx status code (e.g., 402, 403, 409)
return ConsumptionResult(
success=False,
error_code=e.error_code,
error_message=e.error_message
)
except AgentPayConnectionError as e:
# Network error occurred during the request
return ConsumptionResult(
success=False,
error_code="NETWORK_ERROR",
error_message=str(e)
)
except Exception as e:
# Catch any other unexpected errors during the process
return ConsumptionResult(
success=False,
error_code="SDK_ERROR",
error_message=f"An unexpected error occurred in the SDK: {e}"
)
def validate_api_key(self, api_key: str) -> bool:
"""Validates if a User API Key is active and valid for this service.
Args:
api_key: The User's API key to validate.
Returns:
True if the API key is valid for the service, False otherwise.
Returns False if any API or network error occurs during validation.
"""
endpoint = "/api_keys/validate"
payload = {"api_key": api_key}
try:
response_data = self._request("POST", endpoint, data=payload)
# Expecting a response like {"is_valid": true | false}
is_valid = response_data.get("is_valid")
if isinstance(is_valid, bool):
return is_valid
else:
# Unexpected response format from the API
# Log this? For now, treat unexpected format as invalid.
print(f"Warning: Unexpected response format from /validate endpoint: {response_data}")
return False
except (AgentPayAPIError, AgentPayConnectionError) as e:
# If any API or network error occurs, treat the key as invalid for safety.
# Log the error for debugging if needed.
print(f"API or Connection Error during validation: {e}")
return False
except Exception as e:
# Catch any other unexpected errors
print(f"Unexpected SDK error during validation: {e}")
return False |