Infini Payment MCP Server

Instructions

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

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)}