Frappe MCP Server

Instructions

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)}"