Norman Finance MCP Server

Instructions

Implementation Reference

• norman_mcp/tools/documents.py:251-418 (handler)

  Core implementation of create_attachment tool: validates input, downloads URL if needed, uploads file to Norman API with metadata, handles cleanup and errors.
  async def create_attachment( ctx: Context, file_path: str = Field(description="Path or URL to file to upload"), transactions: Optional[List[str]] = Field(description="List of transaction IDs to link"), attachment_type: Optional[str] = Field(description="Type of attachment (invoice, receipt)"), amount: Optional[float] = Field(description="Amount related to attachment"), amount_exchanged: Optional[float] = Field(description="Exchanged amount in different currency"), attachment_number: Optional[str] = Field(description="Unique number for attachment"), brand_name: Optional[str] = Field(description="Brand name associated with attachment"), currency: str = "EUR", currency_exchanged: str = "EUR", description: Optional[str] = Field(description="Description of attachment"), supplier_country: Optional[str] = Field(description="Country of supplier (DE, INSIDE_EU, OUTSIDE_EU)"), value_date: Optional[str] = Field(description="Date of value"), vat_sum_amount: Optional[float] = Field(description="VAT sum amount"), vat_sum_amount_exchanged: Optional[float] = Field(description="Exchanged VAT sum amount"), vat_rate: Optional[int] = Field(description="VAT rate percentage"), sale_type: Optional[str] = Field(description="Type of sale"), additional_metadata: Optional[Dict[str, Any]] = Field(description="Additional metadata for attachment") ) -> Dict[str, Any]: """ Create a new attachment. Args: file_path: Path to file or URL to upload transactions: List of transaction IDs to link attachment_type: Type of attachment (invoice, receipt) amount: Amount related to attachment amount_exchanged: Exchanged amount in different currency attachment_number: Unique number for attachment brand_name: Brand name associated with attachment currency: Currency of amount (default EUR) currency_exchanged: Exchanged currency (default EUR) description: Description of attachment supplier_country: Country of supplier (DE, INSIDE_EU, OUTSIDE_EU) value_date: Date of value vat_sum_amount: VAT sum amount vat_sum_amount_exchanged: Exchanged VAT sum amount vat_rate: VAT rate percentage sale_type: Type of sale additional_metadata: Additional metadata for attachment Returns: Created attachment information """ 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 file path if not validate_file_path(file_path): return {"error": "Invalid or unsafe file path"} # Validate attachment_type if attachment_type and attachment_type not in ["invoice", "receipt", "contract", "other"]: return {"error": "attachment_type must be one of: invoice, receipt, contract, other"} # Validate supplier_country if supplier_country and supplier_country not in ["DE", "INSIDE_EU", "OUTSIDE_EU"]: return {"error": "supplier_country must be one of: DE, INSIDE_EU, OUTSIDE_EU"} # Validate sale_type if sale_type and sale_type not in ["GOODS", "SERVICES"]: return {"error": "sale_type must be one of: GOODS, SERVICES"} attachments_url = urljoin( config.api_base_url, f"api/v1/companies/{company_id}/attachments/" ) try: temp_file_path = None actual_file_path = file_path # Check if file_path is a URL and download it if needed if is_url(file_path): logger.info(f"Downloading file from URL: {file_path}") temp_file_path = download_file(file_path) if not temp_file_path: return {"error": f"Failed to download file from URL: {file_path}"} actual_file_path = temp_file_path logger.info(f"File downloaded to: {actual_file_path}") # Check if file exists and is readable if not os.path.exists(actual_file_path): return {"error": f"File not found: {actual_file_path}"} if not os.access(actual_file_path, os.R_OK): return {"error": f"Permission denied when accessing file: {actual_file_path}"} files = { "file": open(actual_file_path, "rb") } data = {} if transactions: # Validate each transaction ID data["transactions"] = [tx for tx in transactions if validate_input(tx)] if attachment_type: data["attachment_type"] = attachment_type if amount is not None: data["amount"] = amount if amount_exchanged is not None: data["amount_exchanged"] = amount_exchanged if attachment_number: data["attachment_number"] = validate_input(attachment_number) if brand_name: data["brand_name"] = brand_name if currency: data["currency"] = currency if currency_exchanged: data["currency_exchanged"] = currency_exchanged if description: data["description"] = description if supplier_country: data["supplier_country"] = supplier_country if value_date: data["value_date"] = value_date if vat_sum_amount is not None: data["vat_sum_amount"] = vat_sum_amount if vat_sum_amount_exchanged is not None: data["vat_sum_amount_exchanged"] = vat_sum_amount_exchanged if vat_rate is not None: data["vat_rate"] = vat_rate if sale_type: data["sale_type"] = sale_type if additional_metadata: # Sanitize the metadata sanitized_metadata = {} for key, value in additional_metadata.items(): if isinstance(value, str): sanitized_metadata[validate_input(key)] = validate_input(value) else: sanitized_metadata[validate_input(key)] = value data["additional_metadata"] = sanitized_metadata response = api._make_request("POST", attachments_url, json_data=data, files=files) # Close the file handle files["file"].close() # Clean up temporary file if we downloaded one if temp_file_path and os.path.exists(temp_file_path): try: os.remove(temp_file_path) os.rmdir(os.path.dirname(temp_file_path)) logger.info(f"Removed temporary file: {temp_file_path}") except Exception as e: logger.warning(f"Failed to remove temporary file: {str(e)}") return response except FileNotFoundError: return {"error": f"File not found: {file_path}"} except PermissionError: return {"error": f"Permission denied when accessing file: {file_path}"} except Exception as e: # Clean up temporary file if there was an error if temp_file_path and os.path.exists(temp_file_path): try: os.remove(temp_file_path) os.rmdir(os.path.dirname(temp_file_path)) except Exception: pass logger.error(f"Error uploading file: {str(e)}") return {"error": f"Error uploading file: {str(e)}"}
• norman_mcp/tools/documents.py:252-269 (schema)

  Pydantic Field descriptions define the input schema and documentation for the tool parameters.
  ctx: Context, file_path: str = Field(description="Path or URL to file to upload"), transactions: Optional[List[str]] = Field(description="List of transaction IDs to link"), attachment_type: Optional[str] = Field(description="Type of attachment (invoice, receipt)"), amount: Optional[float] = Field(description="Amount related to attachment"), amount_exchanged: Optional[float] = Field(description="Exchanged amount in different currency"), attachment_number: Optional[str] = Field(description="Unique number for attachment"), brand_name: Optional[str] = Field(description="Brand name associated with attachment"), currency: str = "EUR", currency_exchanged: str = "EUR", description: Optional[str] = Field(description="Description of attachment"), supplier_country: Optional[str] = Field(description="Country of supplier (DE, INSIDE_EU, OUTSIDE_EU)"), value_date: Optional[str] = Field(description="Date of value"), vat_sum_amount: Optional[float] = Field(description="VAT sum amount"), vat_sum_amount_exchanged: Optional[float] = Field(description="Exchanged VAT sum amount"), vat_rate: Optional[int] = Field(description="VAT rate percentage"), sale_type: Optional[str] = Field(description="Type of sale"), additional_metadata: Optional[Dict[str, Any]] = Field(description="Additional metadata for attachment")
• norman_mcp/server.py:332-332 (registration)

  Top-level registration by calling register_document_tools on the MCP server instance, which defines and registers the create_attachment tool.
  register_document_tools(server)
• norman_mcp/tools/documents.py:250-250 (registration)

  The @mcp.tool() decorator registers the create_attachment function when register_document_tools is called.
  @mcp.tool()
• 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