Frappe MCP Server

Instructions

    Count documents in Frappe with optional filters.
    
    This tool addresses the filtering limitation that existed in previous implementations
    by using Frappe's native count functionality via the REST API with a custom filter language.
    
    Args:
        doctype: DocType name
        filters: Filter string (optional). Uses custom syntax to bypass MCP validation issues.
    
    Filter Syntax:
        - Simple equality: "field:value" -> {"field": "value"}
        - Operators: "field:operator:value" -> {"field": ["operator", value]}
        - Multiple filters: "field1:value1,field2:operator:value2"
        
    Supported Operators:
        - Equality: = (default), !=
        - Comparison: <, >, <=, >=  
        - Pattern: like, not_like (use % for wildcards)
        - Lists: in, not_in (separate values with |)
        - Null checks: is:null, is:not_null, is_not:null
        - Ranges: between (separate values with |)

    Examples:
        - "status:Unreconciled" -> Status equals Unreconciled
        - "amount:>:100" -> Amount greater than 100
        - "name:like:%admin%" -> Name contains 'admin'
        - "status:in:Open|Working|Pending" -> Status in list
        - "date:between:2025-01-01|2025-12-31" -> Date in range
        - "phone:is:not_null" -> Phone is not null
    
    Tool Examples:
        - count_documents("User") -> Count all users
        - count_documents("Bank Transaction", "status:Unreconciled") -> Count unreconciled transactions
        - count_documents("Bank Transaction", "unallocated_amount:>:0") -> Count with unallocated amount
        - count_documents("Task", "status:in:Open|Working|Pending") -> Count tasks with multiple statuses
        - count_documents("User", "name:like:%admin%") -> Count users with 'admin' in name  
        - count_documents("Payment Entry", "posting_date:between:2025-01-01|2025-12-31") -> Count in date range
        - count_documents("Contact", "phone:is:not_null") -> Count contacts with phone numbers
    

Implementation Reference

• src/tools/documents.py:448-522 (handler)

  The primary handler function for the 'count_documents' MCP tool. Decorated with @mcp.tool() for registration. Parses optional filter strings, queries Frappe REST API /api/resource/{doctype} with count(name) fields and filters, formats and returns the count result or error.

  @mcp.tool()
  async def count_documents(
      doctype: str,
      filters: Optional[str] = None
  ) -> str:
      """
      Count documents in Frappe with optional filters.
      
      This tool addresses the filtering limitation that existed in previous implementations
      by using Frappe's native count functionality via the REST API with a custom filter language.
      
      Args:
          doctype: DocType name
          filters: Filter string (optional). Uses custom syntax to bypass MCP validation issues.
      
      Filter Syntax:
          - Simple equality: "field:value" -> {"field": "value"}
          - Operators: "field:operator:value" -> {"field": ["operator", value]}
          - Multiple filters: "field1:value1,field2:operator:value2"
          
      Supported Operators:
          - Equality: = (default), !=
          - Comparison: <, >, <=, >=  
          - Pattern: like, not_like (use % for wildcards)
          - Lists: in, not_in (separate values with |)
          - Null checks: is:null, is:not_null, is_not:null
          - Ranges: between (separate values with |)
  
      Examples:
          - "status:Unreconciled" -> Status equals Unreconciled
          - "amount:>:100" -> Amount greater than 100
          - "name:like:%admin%" -> Name contains 'admin'
          - "status:in:Open|Working|Pending" -> Status in list
          - "date:between:2025-01-01|2025-12-31" -> Date in range
          - "phone:is:not_null" -> Phone is not null
      
      Tool Examples:
          - count_documents("User") -> Count all users
          - count_documents("Bank Transaction", "status:Unreconciled") -> Count unreconciled transactions
          - count_documents("Bank Transaction", "unallocated_amount:>:0") -> Count with unallocated amount
          - count_documents("Task", "status:in:Open|Working|Pending") -> Count tasks with multiple statuses
          - count_documents("User", "name:like:%admin%") -> Count users with 'admin' in name  
          - count_documents("Payment Entry", "posting_date:between:2025-01-01|2025-12-31") -> Count in date range
          - count_documents("Contact", "phone:is:not_null") -> Count contacts with phone numbers
      """
      try:
          client = get_client()
          
          # Build query parameters for counting
          params = {
              "fields": json.dumps(["count(name) as count"])
          }
          
          # Parse and add filters if provided
          parsed_filters = format_filters_for_api(filters)
          if parsed_filters:
              params["filters"] = json.dumps(parsed_filters)
          
          # Make API request to count documents
          response = await client.get(f"api/resource/{doctype}", params=params)
          
          if "data" in response and response["data"]:
              count_result = response["data"][0]
              count = count_result.get("count", 0)
              
              # Format response based on whether filters were applied
              if parsed_filters:
                  return f"Found {count} {doctype} documents matching filters: {filters}"
              else:
                  return f"Found {count} {doctype} documents total"
          else:
              return f"No data returned for {doctype} count"
              
      except Exception as error:
          return _format_error_response(error, "count_documents")

• src/server.py:39-42 (registration)

  Registration block in create_server() that invokes register_tools(mcp) on all tool modules, including documents.register_tools(mcp) which defines and registers the count_documents handler.

  helpers.register_tools(mcp)
  documents.register_tools(mcp)
  schema.register_tools(mcp)
  reports.register_tools(mcp)

• src/tools/filter_parser.py:11-108 (helper)

  Core helper function parse_filter_string() that implements the custom filter syntax parsing (e.g., 'status:Unreconciled', 'amount:>:100', 'status:in:Open|Working') into Frappe API filter dict format. Called by count_documents via format_filters_for_api() wrapper.

  def parse_filter_string(filter_str: str) -> Dict[str, Any]:
      """
      Parse filter string into Frappe filter format.
      
      Supported operators: =, !=, <, >, <=, >=, like, not_like, in, not_in, is, is_not, between
      
      Examples:
      - "status:Unreconciled" -> {"status": "Unreconciled"}
      - "amount:>:100" -> {"amount": [">", 100]}
      - "name:like:%test%" -> {"name": ["like", "%test%"]}
      - "status:in:Open|Closed" -> {"status": ["in", ["Open", "Closed"]]}
      - "date:between:2025-01-01|2025-12-31" -> {"date": ["between", ["2025-01-01", "2025-12-31"]]}
      - "field:is:null" -> {"field": ["is", "not set"]}
      - "date:>=:2024-01-01,date:<=:2024-01-31" -> {"date": [[">=", "2024-01-01"], ["<=", "2024-01-31"]]}
      """
      filters_dict: Dict[str, Any] = {}
      
      # Handle multiple filters separated by commas
      filter_parts = filter_str.split(',')
      
      for part in filter_parts:
          part = part.strip()
          if ':' in part:
              # Split on first two colons to handle operators with underscores
              components = part.split(':', 2)
              
              if len(components) >= 3:
                  # Format: field:operator:value(s)
                  field, operator, value_str = components[0].strip(), components[1].strip(), components[2]
                  
                  # Create the filter condition
                  filter_condition = None
                  
                  # Handle special operators
                  if operator.lower() in ['in', 'not_in']:
                      # Handle list values separated by |
                      values = [v.strip() for v in value_str.split('|')]
                      # Convert numbers in list
                      converted_values = []
                      for v in values:
                          converted_values.append(_convert_value(v))
                      filter_condition = [operator.replace('_', ' '), converted_values]
                      
                  elif operator.lower() == 'between':
                      # Handle range values separated by |
                      range_values = [v.strip() for v in value_str.split('|')]
                      if len(range_values) == 2:
                          converted_range = [_convert_value(v) for v in range_values]
                          filter_condition = [operator, converted_range]
                      else:
                          raise ValueError(f"Between operator requires exactly 2 values separated by |, got: {value_str}")
                          
                  elif operator.lower() in ['is', 'is_not']:
                      # Handle null checks: is:null, is:not_null, is_not:null, etc.
                      if value_str.lower() in ['null', 'none', 'empty']:
                          filter_condition = [operator.replace('_', ' '), "not set"]
                      elif value_str.lower() in ['not_null', 'not_none', 'not_empty']:
                          filter_condition = [operator.replace('_', ' '), "set"]
                      else:
                          filter_condition = [operator.replace('_', ' '), _convert_value(value_str)]
                          
                  elif operator.lower() == 'not_like':
                      # Handle not like operator  
                      filter_condition = ["not like", value_str]
                      
                  else:
                      # Standard operators: =, !=, <, >, <=, >=, like
                      filter_condition = [operator, _convert_value(value_str)]
                  
                  # Handle multiple filters for the same field
                  if field in filters_dict:
                      # Convert existing single filter to list format
                      if not isinstance(filters_dict[field], list) or len(filters_dict[field]) != 2 or not isinstance(filters_dict[field][0], list):
                          filters_dict[field] = [filters_dict[field]]
                      # Add new filter condition
                      filters_dict[field].append(filter_condition)
                  else:
                      filters_dict[field] = filter_condition
                      
              elif len(components) == 2:
                  # Simple field:value format (implies equality)
                  field, value_str = components[0].strip(), components[1]
                  filter_condition = _convert_value(value_str)
                  
                  # Handle multiple filters for the same field
                  if field in filters_dict:
                      # Convert existing single filter to list format
                      if not isinstance(filters_dict[field], list) or len(filters_dict[field]) != 2 or not isinstance(filters_dict[field][0], list):
                          filters_dict[field] = [filters_dict[field]]
                      # Add new filter condition
                      filters_dict[field].append(filter_condition)
                  else:
                      filters_dict[field] = filter_condition
                  
      # Post-process: Convert >= and <= on same field to between operator
      filters_dict = _optimize_range_filters(filters_dict)
      
      return filters_dict

• src/tools/documents.py:51-79 (helper)

  Shared helper function _format_error_response() used by count_documents (and other document tools) to provide consistent, diagnostic-rich error messages including credential checks and Frappe-specific handling.

  def _format_error_response(error: Exception, operation: str) -> str:
      """Format error response with detailed information."""
      credentials_check = validate_api_credentials()
      
      # Build diagnostic information
      diagnostics = [
          f"Error in {operation}",
          f"Error type: {type(error).__name__}",
          f"Is FrappeApiError: {isinstance(error, FrappeApiError)}",
          f"API Key available: {credentials_check['details']['api_key_available']}",
          f"API Secret available: {credentials_check['details']['api_secret_available']}"
      ]
      
      # Check for missing credentials first
      if not credentials_check["valid"]:
          error_msg = f"Authentication failed: {credentials_check['message']}. "
          error_msg += "API key/secret is the only supported authentication method."
          return error_msg
      
      # Handle FrappeApiError
      if isinstance(error, FrappeApiError):
          error_msg = f"Frappe API error: {error}"
          if error.status_code in (401, 403):
              error_msg += " Please check your API key and secret."
          return error_msg
      
      # Default error handling
      return f"Error in {operation}: {str(error)}"