Norman Finance MCP Server

Instructions

Create a new attachment.

Args:
    file_path: Path to file 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

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