Norman Finance MCP Server

Instructions

Upload multiple file attachments in bulk.

Args:
    file_paths: List of paths to files to upload
    cashflow_type: Optional cashflow type for the transactions (INCOME or EXPENSE)
    
Returns:
    Response from the bulk upload request

Implementation Reference

• norman_mcp/tools/documents.py:81-202 (handler)

  Core handler implementation for upload_bulk_attachments tool. Processes list of file paths or URLs, downloads URLs to temp files if needed, validates paths, opens files, sends multipart POST to API endpoint /api/v1/accounting/transactions/upload-documents/, handles cleanup and errors.

  @mcp.tool()
  async def upload_bulk_attachments(
      ctx: Context,
      file_paths: List[str] = Field(description="List of paths or URLs to files to upload"),
      cashflow_type: Optional[str] = Field(description="Optional cashflow type for the transactions (INCOME or EXPENSE). If not provided, then try to detect it from the file")
  ) -> Dict[str, Any]:
      """
      Upload multiple file attachments in bulk.
      
      Args:
          file_paths: List of paths or URLs to files to upload
          cashflow_type: Optional cashflow type for the transactions (INCOME or EXPENSE). If not provided, then try to detect it from the file
          
      Returns:
          Response from the bulk upload request
      """
      api = ctx.request_context.lifespan_context["api"]
      company_id = api.company_id
      
      if not company_id:
          return {"error": "No company available. Please authenticate first."}
      
      # Validate cashflow_type
      if cashflow_type and cashflow_type not in ["INCOME", "EXPENSE"]:
          return {"error": "cashflow_type must be either 'INCOME' or 'EXPENSE'"}
          
      upload_url = urljoin(
          config.api_base_url,
          "api/v1/accounting/transactions/upload-documents/"
      )
      
      temp_files = []  # Track temp files for cleanup
      opened_files = []  # Track opened file handles for cleanup
      
      try:
          files = []
          valid_paths = []
          
          # Process all file paths before proceeding
          for path in file_paths:
              if not validate_file_path(path):
                  logger.warning(f"Invalid or unsafe file path: {path}")
                  continue
              
              actual_path = path
              
              # Handle URLs by downloading them first
              if is_url(path):
                  logger.info(f"Downloading file from URL: {path}")
                  downloaded_path = download_file(path)
                  if not downloaded_path:
                      logger.warning(f"Failed to download file from URL: {path}")
                      continue
                  actual_path = downloaded_path
                  temp_files.append(actual_path)
                  logger.info(f"File downloaded to: {actual_path}")
              
              # Validate the file exists and is accessible
              if not os.path.exists(actual_path):
                  logger.warning(f"File not found: {actual_path}")
                  continue
              
              if not os.access(actual_path, os.R_OK):
                  logger.warning(f"Permission denied when accessing file: {actual_path}")
                  continue
                  
              valid_paths.append(actual_path)
              
          if not valid_paths:
              return {"error": "No valid files found for upload"}
              
          # Open and prepare valid files
          for path in valid_paths:
              file_handle = open(path, "rb")
              opened_files.append(file_handle)
              files.append(("files", file_handle))
                  
          data = {}
          if cashflow_type:
              data["cashflow_type"] = cashflow_type
              
          response = api._make_request("POST", upload_url, json_data=data, files=files)
          
          # Close all opened file handles
          for file_handle in opened_files:
              file_handle.close()
              
          # Clean up temporary files
          for temp_file in temp_files:
              try:
                  if os.path.exists(temp_file):
                      os.remove(temp_file)
                      os.rmdir(os.path.dirname(temp_file))
                      logger.info(f"Removed temporary file: {temp_file}")
              except Exception as e:
                  logger.warning(f"Failed to remove temporary file {temp_file}: {str(e)}")
                  
          return response
          
      except FileNotFoundError as e:
          return {"error": f"File not found: {str(e)}"}
      except PermissionError as e:
          return {"error": f"Permission denied when accessing file: {str(e)}"}
      except Exception as e:
          logger.error(f"Error uploading files: {str(e)}")
          return {"error": f"Error uploading files: {str(e)}"}
      finally:
          # Ensure files are closed and temp files are cleaned up in case of exceptions
          for file_handle in opened_files:
              try:
                  file_handle.close()
              except Exception:
                  pass
                  
          for temp_file in temp_files:
              try:
                  if os.path.exists(temp_file):
                      os.remove(temp_file)
                      os.rmdir(os.path.dirname(temp_file))
              except Exception:
                  pass

• norman_mcp/server.py:332-332 (registration)

  Top-level registration of document tools (including upload_bulk_attachments) by calling register_document_tools(server) in the main server setup.

  register_document_tools(server)

• norman_mcp/server.py:20-20 (registration)

  Import of register_document_tools function required for tool registration.

  from norman_mcp.tools.documents import register_document_tools

• norman_mcp/tools/documents.py:82-86 (schema)

  Pydantic schema definition for tool parameters using Field with descriptions.

  async def upload_bulk_attachments(
      ctx: Context,
      file_paths: List[str] = Field(description="List of paths or URLs to files to upload"),
      cashflow_type: Optional[str] = Field(description="Optional cashflow type for the transactions (INCOME or EXPENSE). If not provided, then try to detect it from the file")
  ) -> Dict[str, Any]:

• norman_mcp/tools/documents.py:25-58 (helper)

  Helper function to download files from URLs to temporary local paths for upload.

  def download_file(url: str) -> Optional[str]:
      """Download a file from URL to a temporary location and return its path."""
      try:
          response = requests.get(url, stream=True, timeout=30)
          response.raise_for_status()
          
          # Extract filename from URL or Content-Disposition header
          filename = None
          
          if "Content-Disposition" in response.headers:
              # Try to get filename from Content-Disposition header
              content_disposition = response.headers["Content-Disposition"]
              match = re.search(r'filename="?([^"]+)"?', content_disposition)
              if match:
                  filename = match.group(1)
          
          # If no filename found in header, extract from URL
          if not filename:
              url_path = urlparse(url).path
              filename = os.path.basename(url_path) or "downloaded_file"
          
          # Create a temporary file
          temp_dir = tempfile.mkdtemp(prefix="norman_")
          temp_path = os.path.join(temp_dir, filename)
          
          # Write the file
          with open(temp_path, 'wb') as f:
              for chunk in response.iter_content(chunk_size=8192):
                  f.write(chunk)
                  
          return temp_path
      except Exception as e:
          logger.error(f"Error downloading file from {url}: {str(e)}")
          return None