natural_query
Processes natural language queries to retrieve structured data about leads, contacts, and opportunities from Ambivo MCP Server, enabling efficient data extraction in table, natural, or both formats.
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 implementation of the natural_query tool logic in AmbivoAPIClient. Makes POST request to /entity/natural_query with validation, retries, and error handling.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:176-200 (registration)Registration of the natural_query tool in the MCP server's list_tools handler, including its description and input schema.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:269-305 (handler)MCP server tool call handler dispatch for natural_query, which validates auth and arguments then invokes the core natural_query implementation.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:182-199 (schema)Input schema definition for the natural_query tool, specifying parameters query (required) and response_format."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"], },