Infini Payment MCP Server

Instructions

Implementation Reference

• server.py:130-173 (handler)

  The handler function that implements the create_payment_order tool logic. It constructs the request data from parameters and calls the InfiniClient to POST to /order endpoint.
  def create_payment_order( request_id: str, amount: str, client_reference: Optional[str] = None, order_desc: Optional[str] = None, merchant_alias: Optional[str] = None, expires_in: Optional[int] = 0, success_url: Optional[str] = None, failure_url: Optional[str] = None ) -> str: """ Create a new payment order in Infini. Args: request_id: Unique request ID (UUID recommended) amount: Order amount (e.g., "100.00") client_reference: Client reference string order_desc: Order description merchant_alias: Custom merchant alias expires_in: Time to live in seconds (0 for default) success_url: Success redirect URL failure_url: Failure redirect URL """ data = { "request_id": request_id, "amount": amount } # Add optional fields if client_reference: data["client_reference"] = client_reference if order_desc: data["order_desc"] = order_desc if merchant_alias: data["merchant_alias"] = merchant_alias if expires_in is not None: data["expires_in"] = expires_in if success_url: data["success_url"] = success_url if failure_url: data["failure_url"] = failure_url result = client.request("POST", "/order", json_data=data) return str(result)
• server.py:129-129 (registration)

  Registers the create_payment_order function as an MCP tool using the FastMCP decorator.
  @mcp.tool()
• server.py:130-139 (schema)

  Function signature defines the input schema via type annotations and defaults, output is str (JSON response).
  def create_payment_order( request_id: str, amount: str, client_reference: Optional[str] = None, order_desc: Optional[str] = None, merchant_alias: Optional[str] = None, expires_in: Optional[int] = 0, success_url: Optional[str] = None, failure_url: Optional[str] = None ) -> str:
• server.py:27-126 (helper)

  InfiniClient class provides the HTTP client with authentication (HMAC signing) used by the tool to interact with the Infini API.
  class InfiniClient: def __init__(self): self.key_id = INFINI_API_KEY self.secret_key = INFINI_SECRET_KEY.encode() if INFINI_SECRET_KEY else None self.base_url = BASE_URLS.get(INFINI_ENV, BASE_URLS["sandbox"]) if not self.key_id or not self.secret_key: print("Warning: INFINI_API_KEY or INFINI_SECRET_KEY not set.") def _sign_request(self, method: str, path: str, body: str = None) -> Dict[str, str]: """ Generate HMAC-SHA256 signature based on Infini documentation. Args: method: HTTP method (GET, POST, etc.) path: API path (e.g., /order) body: JSON body string for POST requests Returns: Dict of headers including Authorization """ gmt_time = datetime.now(timezone.utc).strftime('%a, %d %b %Y %H:%M:%S GMT') # Build signing string signing_string = f"{self.key_id}\n{method.upper()} {path}\ndate: {gmt_time}\n" body_digest_base64 = None if body is not None: # Calculate body digest body_digest = hashlib.sha256(body.encode('utf-8')).digest() body_digest_base64 = base64.b64encode(body_digest).decode('utf-8') signing_string += f"digest: SHA-256={body_digest_base64}\n" # Calculate signature signature = base64.b64encode( hmac.new(self.secret_key, signing_string.encode(), hashlib.sha256).digest() ).decode() # Build headers headers = { "Date": gmt_time, "Authorization": f'Signature keyId="{self.key_id}",algorithm="hmac-sha256",' f'headers="@request-target date",signature="{signature}"' } if body_digest_base64 is not None: headers["Digest"] = f"SHA-256={body_digest_base64}" return headers def request(self, method: str, path: str, json_data: Dict = None, params: Dict = None) -> Dict: """ Make HTTP request to Infini API with proper authentication. Args: method: HTTP method path: API path json_data: Request body for POST requests params: Query parameters for GET requests Returns: Response data or error dict """ if not self.key_id or not self.secret_key: return {"error": "API credentials not configured"} url = f"{self.base_url}{path}" # Prepare body and headers body = None if json_data is not None: body = json.dumps(json_data, separators=(',', ':')) headers = self._sign_request(method, path, body) if json_data is not None: headers["Content-Type"] = "application/json" try: with httpx.Client() as client: if method.upper() == "GET" and params: url += f"?{urlencode(params)}" response = client.request( method, url, data=body, headers=headers, timeout=30.0 ) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: return { "error": f"HTTP Error: {e.response.status_code}", "detail": e.response.text } except Exception as e: return {"error": str(e)}