natural_query
Process natural language queries to retrieve structured data about leads, contacts, opportunities, and other entities from Ambivo's CRM system.
Instructions
Execute natural language queries against Ambivo entity data. This tool processes natural language queries and returns structured data about leads, contacts, opportunities, and other entities.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | Natural language query describing what data you want to retrieve. Examples: 'Show me leads created this week', 'Find contacts with gmail addresses', 'List opportunities worth more than $10,000' | |
| response_format | No | Format of the response: 'table' for structured data, 'natural' for natural language description, 'both' for both formats | both |
Implementation Reference
- ambivo_mcp_server/server.py:107-160 (handler)Core handler function in AmbivoAPIClient that implements the natural_query tool logic: validates inputs, makes authenticated POST request to /entity/natural_query endpoint with retry logic, handles errors, and returns the API response.async def natural_query( self, query: str, response_format: str = "both" ) -> Dict[str, Any]: """ Execute a natural language query against entity data with validation and error handling Args: query: Natural language query string response_format: Response format - "table", "natural", or "both" Returns: API response dictionary """ # Validate inputs input_validator.validate_query(query) if response_format not in ["table", "natural", "both"]: raise ValueError( "Invalid response_format. Must be 'table', 'natural', or 'both'" ) payload = {"query": query, "response_format": response_format} url = f"{self.base_url}/entity/natural_query" try: self.logger.info(f"Executing natural query: {query[:100]}...") start_time = time.time() response = await self._make_request_with_retry( "POST", url, json=payload, headers=self._get_headers() ) elapsed_time = time.time() - start_time self.logger.info(f"Natural query completed in {elapsed_time:.2f}s") response.raise_for_status() result = response.json() self.logger.debug(f"API response: {json.dumps(result, indent=2)[:500]}...") return result except httpx.TimeoutException as e: self.logger.error(f"Natural query timeout: {e}") raise Exception(f"Request timeout after {self.config.timeout}s") except httpx.HTTPStatusError as e: self.logger.error( f"Natural query HTTP error: {e.response.status_code} - {e.response.text}" ) raise except Exception as e: self.logger.error(f"Natural query unexpected error: {e}") raise
- ambivo_mcp_server/server.py:269-305 (handler)MCP server tool call handler dispatch block for 'natural_query': checks authentication and arguments, invokes the core natural_query method, formats the result as TextContent, and handles errors.elif name == "natural_query": if not api_client.auth_token: return [ types.TextContent( type="text", text="Error: Authentication required. Please use the 'set_auth_token' tool first.", ) ] query = arguments.get("query") if not query: return [ types.TextContent( type="text", text="Error: Query parameter is required" ) ] response_format = arguments.get("response_format", "both") try: result = await api_client.natural_query(query, response_format) return [ types.TextContent( type="text", text=f"Natural Query Results:\n\n{json.dumps(result, indent=2)}", ) ] except httpx.HTTPStatusError as e: error_msg = f"HTTP {e.response.status_code}: {e.response.text}" return [types.TextContent(type="text", text=f"API Error: {error_msg}")] except Exception as e: return [ types.TextContent( type="text", text=f"Error executing natural query: {str(e)}" ) ]
- ambivo_mcp_server/server.py:176-200 (registration)Tool registration in @server.list_tools(): defines the 'natural_query' tool with name, description, and input schema for MCP protocol.types.Tool( name="natural_query", description="Execute natural language queries against Ambivo entity data. " "This tool processes natural language queries and returns structured data " "about leads, contacts, opportunities, and other entities.", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "Natural language query describing what data you want to retrieve. " "Examples: 'Show me leads created this week', 'Find contacts with gmail addresses', " "'List opportunities worth more than $10,000'", }, "response_format": { "type": "string", "enum": ["table", "natural", "both"], "default": "both", "description": "Format of the response: 'table' for structured data, " "'natural' for natural language description, 'both' for both formats", }, }, "required": ["query"], }, ),
- ambivo_mcp_server/server.py:181-199 (schema)Input schema for the natural_query tool: requires 'query' string, optional 'response_format' with enum values.inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "Natural language query describing what data you want to retrieve. " "Examples: 'Show me leads created this week', 'Find contacts with gmail addresses', " "'List opportunities worth more than $10,000'", }, "response_format": { "type": "string", "enum": ["table", "natural", "both"], "default": "both", "description": "Format of the response: 'table' for structured data, " "'natural' for natural language description, 'both' for both formats", }, }, "required": ["query"], },