#!/usr/bin/env python3 """ Teleport Pay MCP Server Connect AI assistants (Claude, Cursor, etc.) to your Teleport Pay account. Usage: python teleport-pay-server.py Environment variables: TELEPORT_PAY_API_KEY — your tk_... API key (required) TELEPORT_PAY_BASE_URL — API base URL (default: https://gate.teleport.ltd/api/crypto-acquiring/v1) """ import os import sys import json import urllib.request import urllib.error from typing import Any # --------------------------------------------------------------------------- # MCP protocol helpers (stdio JSON-RPC transport) # --------------------------------------------------------------------------- def _read_message() -> dict | None: """Read a JSON-RPC message from stdin (Content-Length framing).""" headers: dict[str, str] = {} while True: line = sys.stdin.buffer.readline() if not line: return None line_str = line.decode("utf-8").rstrip("\r\n") if line_str == "": break if ":" in line_str: key, value = line_str.split(":", 1) headers[key.strip().lower()] = value.strip() length = int(headers.get("content-length", 0)) if length == 0: return None body = sys.stdin.buffer.read(length) return json.loads(body.decode("utf-8")) def _send_message(msg: dict): """Write a JSON-RPC message to stdout (Content-Length framing).""" body = json.dumps(msg).encode("utf-8") sys.stdout.buffer.write(f"Content-Length: {len(body)}\r\n\r\n".encode("utf-8")) sys.stdout.buffer.write(body) sys.stdout.buffer.flush() def _result(id: Any, result: Any): _send_message({"jsonrpc": "2.0", "id": id, "result": result}) def _error(id: Any, code: int, message: str): _send_message({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}}) # --------------------------------------------------------------------------- # Teleport Pay API client # --------------------------------------------------------------------------- API_KEY = os.environ.get("TELEPORT_PAY_API_KEY", "") BASE_URL = os.environ.get( "TELEPORT_PAY_BASE_URL", "https://gate.teleport.ltd/api/crypto-acquiring/v1", ).rstrip("/") def _api_request(method: str, path: str, body: dict | None = None) -> dict: """Make an HTTP request to the Teleport Pay API.""" url = f"{BASE_URL}{path}" data = json.dumps(body).encode("utf-8") if body else None req = urllib.request.Request(url, data=data, method=method) req.add_header("X-API-Key", API_KEY) if data: req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: body_text = e.read().decode("utf-8", errors="replace") try: return json.loads(body_text) except Exception: return {"success": False, "error": f"HTTP {e.code}: {body_text[:500]}"} except Exception as exc: return {"success": False, "error": str(exc)} # --------------------------------------------------------------------------- # Tool definitions # --------------------------------------------------------------------------- TOOLS = [ { "name": "list_payment_methods", "description": "List available payment method groups and their methods for the authenticated account.", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "list_currencies", "description": "List available crypto and fiat currencies for invoicing.", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "create_invoice", "description": "Create a payment invoice. Returns a payment URL for the customer.", "inputSchema": { "type": "object", "properties": { "amount": {"type": "number", "description": "Payment amount (e.g. 49.99)"}, "currency": {"type": "string", "description": "Currency code (USD, EUR, RUB, USDT, etc.)"}, "order_id": {"type": "string", "description": "Your internal order ID"}, "description": {"type": "string", "description": "Payment description shown to customer"}, "customer_email": {"type": "string", "description": "Customer email"}, "customer_name": {"type": "string", "description": "Customer name"}, "return_url": {"type": "string", "description": "Redirect URL after successful payment"}, "cancel_url": {"type": "string", "description": "Redirect URL if customer cancels"}, "webhook_url": {"type": "string", "description": "Webhook URL for status notifications"}, "payment_methods": { "type": "array", "items": {"type": "string"}, "description": "Allowed payment method groups (e.g. [\"crypto\", \"cards\"]). If omitted, all available methods are enabled.", }, }, "required": ["amount", "currency"], }, }, { "name": "get_key_info", "description": "Get current API key settings (name, webhook URL, status).", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "update_key_settings", "description": "Update API key settings (webhook URL and/or name).", "inputSchema": { "type": "object", "properties": { "webhook_url": {"type": "string", "description": "New default webhook URL"}, "name": {"type": "string", "description": "New key label/name"}, }, }, }, ] # --------------------------------------------------------------------------- # Tool handlers # --------------------------------------------------------------------------- def handle_tool(name: str, arguments: dict) -> list[dict]: """Execute a tool and return MCP content blocks.""" if name == "list_payment_methods": result = _api_request("GET", "/payment-methods/") elif name == "list_currencies": result = _api_request("GET", "/currencies/") elif name == "create_invoice": body = {} for key in ("amount", "currency", "order_id", "description", "customer_email", "customer_name", "return_url", "cancel_url", "webhook_url", "payment_methods"): if key in arguments and arguments[key] is not None: body[key] = arguments[key] result = _api_request("POST", "/invoice/create/", body) elif name == "get_key_info": result = _api_request("GET", "/key/info/") elif name == "update_key_settings": body = {} if "webhook_url" in arguments: body["webhook_url"] = arguments["webhook_url"] if "name" in arguments: body["name"] = arguments["name"] result = _api_request("PATCH", "/key/update/", body) else: return [{"type": "text", "text": f"Unknown tool: {name}"}] return [{"type": "text", "text": json.dumps(result, indent=2, ensure_ascii=False)}] # --------------------------------------------------------------------------- # Main loop # --------------------------------------------------------------------------- def main(): if not API_KEY: sys.stderr.write("Error: TELEPORT_PAY_API_KEY environment variable is required.\n") sys.exit(1) while True: msg = _read_message() if msg is None: break method = msg.get("method", "") msg_id = msg.get("id") params = msg.get("params", {}) if method == "initialize": _result(msg_id, { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": { "name": "teleport-pay", "version": "1.0.0", }, }) elif method == "notifications/initialized": pass # no response needed elif method == "tools/list": _result(msg_id, {"tools": TOOLS}) elif method == "tools/call": tool_name = params.get("name", "") arguments = params.get("arguments", {}) try: content = handle_tool(tool_name, arguments) _result(msg_id, {"content": content}) except Exception as exc: _result(msg_id, { "content": [{"type": "text", "text": f"Error: {exc}"}], "isError": True, }) elif msg_id is not None: _error(msg_id, -32601, f"Method not found: {method}") if __name__ == "__main__": main()