ydb_query_with_params
Execute parameterized SQL queries using JSON parameters to interact securely and efficiently with YDB databases via the MCP server.
Instructions
Run a parameterized SQL query with JSON parameters
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| params | Yes | ||
| sql | Yes |
Implementation Reference
- ydb_mcp/server.py:504-532 (handler)Main handler function for 'ydb_query_with_params' tool. Parses JSON parameters string into YDB parameters and delegates to the generic query method.async def query_with_params(self, sql: str, params: str) -> List[TextContent]: """Run a parameterized SQL query with JSON parameters. Args: sql: SQL query to execute params: Parameters as a JSON string Returns: Query results as a list of TextContent objects or a dictionary """ # Handle authentication errors if self.auth_error: logger.error(f"Authentication error: {self.auth_error}") safe_error = self._stringify_dict_keys({"error": f"Authentication error: {self.auth_error}"}) return [TextContent(type="text", text=json.dumps(safe_error, indent=2))] try: ydb_params = self._parse_str_to_ydb_params(params) return await self.query(sql, ydb_params) except json.JSONDecodeError as e: logger.error(f"Error parsing JSON parameters: {str(e)}") safe_error = self._stringify_dict_keys({"error": f"Error parsing JSON parameters: {str(e)}"}) return [TextContent(type="text", text=json.dumps(safe_error, indent=2))] except Exception as e: error_message = f"Error executing parameterized query: {str(e)}" logger.error(error_message) safe_error = self._stringify_dict_keys({"error": error_message}) return [TextContent(type="text", text=json.dumps(safe_error, indent=2))]
- ydb_mcp/server.py:599-611 (registration)Tool registration specification including name, description, handler reference, and input parameters schema.{ "name": "ydb_query_with_params", "description": "Run a parameterized SQL query with JSON parameters", "handler": self.query_with_params, # Use real handler "parameters": { "properties": { "sql": {"type": "string", "title": "Sql"}, "params": {"type": "string", "title": "Params"}, }, "required": ["sql", "params"], "type": "object", }, },
- ydb_mcp/server.py:603-610 (schema)Input schema definition for the tool, specifying sql and params as required string parameters."parameters": { "properties": { "sql": {"type": "string", "title": "Sql"}, "params": {"type": "string", "title": "Params"}, }, "required": ["sql", "params"], "type": "object", },
- ydb_mcp/server.py:533-553 (helper)Helper function to parse JSON string parameters into YDB-compatible parameter dictionary, handling typed parameters.def _parse_str_to_ydb_params(self, params: str) -> Dict: parsed_params = {} if params and params.strip(): parsed_params = json.loads(params) # Convert [value, type] to YDB type if needed ydb_params = {} for key, value in parsed_params.items(): param_key = key if key.startswith("$") else f"${key}" if isinstance(value, (list, tuple)) and len(value) == 2: param_value, type_name = value if isinstance(type_name, str) and hasattr(ydb.PrimitiveType, type_name): ydb_type = getattr(ydb.PrimitiveType, type_name) ydb_params[param_key] = (param_value, ydb_type) else: ydb_params[param_key] = param_value else: ydb_params[param_key] = value return ydb_params
- ydb_mcp/server.py:1062-1063 (handler)Dispatch logic in call_tool method that routes requests for 'ydb_query_with_params' to the handler.elif tool_name == "ydb_query_with_params" and "sql" in params and "params" in params: result = await self.query_with_params(sql=params["sql"], params=params["params"])