Norman Finance MCP Server

Instructions

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