""" Tayapro Communications Cloud - Python SDK from tayapro import Tayapro tayapro = Tayapro(api_key=os.environ["TAYAPRO_API_KEY"], base_url="https://") message = tayapro.messages.send(to="+12025550123", body="Your appointment is confirmed.") call = tayapro.calls.create(to="+12025550123") Standard library only (Python 3.9+), so it drops into any environment with no dependencies. Keep the API key server-side. """ from __future__ import annotations import hashlib import hmac import json import time import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, Optional __all__ = ["Tayapro", "TayaproError", "verify_webhook_signature"] USER_AGENT = "tayapro-python/1.0" class TayaproError(Exception): def __init__(self, message: str, status: int = 0, code: str = "request_failed", body: Any = None): super().__init__(message) self.status = status self.code = code self.body = body @property def retryable(self) -> bool: """Rate limits and server faults are the only failures worth retrying.""" return self.status == 429 or self.status >= 500 class _Messages: def __init__(self, client: "Tayapro"): self._client = client def send(self, to: str, body: str, from_: Optional[str] = None, channel: str = "sms") -> Dict[str, Any]: payload: Dict[str, Any] = {"to": to, "body": body, "channel": channel} if from_: payload["from"] = from_ return self._client.request("POST", "/api/public/v1/messages", payload) def list(self, limit: int = 50) -> Dict[str, Any]: return self._client.request("GET", "/api/public/v1/messages", query={"limit": limit}) def get(self, message_id: str) -> Dict[str, Any]: return self._client.request("GET", f"/api/public/v1/messages/{urllib.parse.quote(message_id)}") class _Calls: def __init__(self, client: "Tayapro"): self._client = client def create(self, to: str, from_: Optional[str] = None, answer_url: Optional[str] = None) -> Dict[str, Any]: payload: Dict[str, Any] = {"to": to} if from_: payload["from"] = from_ if answer_url: payload["answer_url"] = answer_url return self._client.request("POST", "/api/public/v1/calls", payload) def list(self, limit: int = 50) -> Dict[str, Any]: return self._client.request("GET", "/api/public/v1/calls", query={"limit": limit}) def get(self, call_id: str) -> Dict[str, Any]: return self._client.request("GET", f"/api/public/v1/calls/{urllib.parse.quote(call_id)}") class _Numbers: def __init__(self, client: "Tayapro"): self._client = client def list(self) -> Dict[str, Any]: return self._client.request("GET", "/api/public/v1/numbers") def available(self, country: str = "US", area_code: Optional[str] = None, contains: Optional[str] = None) -> Dict[str, Any]: query: Dict[str, Any] = {"available": "true", "country": country} if area_code: query["area_code"] = area_code if contains: query["contains"] = contains return self._client.request("GET", "/api/public/v1/numbers", query=query) def buy(self, phone_number: str, label: Optional[str] = None) -> Dict[str, Any]: payload: Dict[str, Any] = {"phone_number": phone_number} if label: payload["label"] = label return self._client.request("POST", "/api/public/v1/numbers", payload) class _Verify: def __init__(self, client: "Tayapro"): self._client = client def start(self, to: str, service: Optional[str] = None) -> Dict[str, Any]: payload: Dict[str, Any] = {"to": to} if service: payload["service"] = service return self._client.request("POST", "/api/public/v1/verify", payload) def check(self, to: str, code: str) -> Dict[str, Any]: return self._client.request("POST", "/api/public/v1/verify/check", {"to": to, "code": code}) class Tayapro: def __init__(self, api_key: str, base_url: str, max_retries: int = 2, timeout: float = 30.0): if not api_key: raise TayaproError("An API key is required (create one under Team & API keys).") if not base_url: raise TayaproError("Pass the base URL of your Tayapro app.") self.api_key = api_key self.base_url = base_url.rstrip("/") self.max_retries = max_retries self.timeout = timeout self.messages = _Messages(self) self.calls = _Calls(self) self.numbers = _Numbers(self) self.verify = _Verify(self) def request( self, method: str, path: str, payload: Optional[Dict[str, Any]] = None, query: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: url = self.base_url + path if query: clean = {key: value for key, value in query.items() if value is not None} url = f"{url}?{urllib.parse.urlencode(clean)}" data = json.dumps(payload).encode() if payload is not None else None last_error: Optional[TayaproError] = None for attempt in range(self.max_retries + 1): request = urllib.request.Request( url, data=data, method=method, headers={ "authorization": f"Bearer {self.api_key}", "content-type": "application/json", "user-agent": USER_AGENT, }, ) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: body = response.read().decode() or "{}" return json.loads(body) except urllib.error.HTTPError as http_error: raw = http_error.read().decode() or "{}" try: parsed = json.loads(raw) except json.JSONDecodeError: parsed = {"raw": raw} error = TayaproError( parsed.get("error", f"Request failed with {http_error.code}"), status=http_error.code, code=parsed.get("error", "request_failed"), body=parsed, ) if not error.retryable or attempt == self.max_retries: raise error last_error = error time.sleep(0.5 * (2**attempt)) except urllib.error.URLError as url_error: last_error = TayaproError(f"Network error: {url_error.reason}") if attempt == self.max_retries: raise last_error time.sleep(0.4 * (attempt + 1)) raise last_error or TayaproError("Request failed") def verify_webhook_signature(header: str, body: str, secret: str, tolerance_seconds: int = 300) -> bool: """Verify x-tayapro-signature: t=,v1= on an inbound webhook.""" parts = dict( piece.strip().split("=", 1) for piece in (header or "").split(",") if "=" in piece ) try: timestamp = int(parts.get("t", "0")) except ValueError: return False signature = parts.get("v1", "") if not timestamp or not signature: return False if abs(time.time() - timestamp) > tolerance_seconds: return False expected = hmac.new( secret.encode(), f"{timestamp}.{body}".encode(), hashlib.sha256 ).hexdigest() return hmac.compare_digest(expected, signature)