ydb_query_with_params
Run parameterized SQL queries on YDB databases by providing SQL and JSON parameters. Enables safe, dynamic database operations from MCP-powered AI assistants.
Instructions
Run a parameterized SQL query with JSON parameters
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes | ||
| params | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- ydb_mcp/tools.py:41-47 (handler)The async function that executes the 'ydb_query_with_params' tool logic. It receives 'sql' (str) and 'params' (str|dict), parses the parameters via _parse_params_str, executes the query via server.execute(), and serializes the result. Returns TextContent with JSON result or error.
async def ydb_query_with_params(sql: str, params: str | dict) -> list[TextContent]: """Run a parameterized SQL query with JSON parameters.""" try: result_sets = await server.execute(sql, _parse_params_str(params)) return [TextContent(type="text", text=serialize_ydb_response({"result_sets": result_sets}))] except Exception as e: return [TextContent(type="text", text=json.dumps({"error": str(e)}, indent=2))] - ydb_mcp/tools.py:20-20 (schema)Enum definition of YDBGenericTool.QUERY_WITH_PARAMS = 'ydb_query_with_params', the constant used to identify this tool.
QUERY_WITH_PARAMS = "ydb_query_with_params" - ydb_mcp/params.py:24-32 (helper)The _parse_params_str helper function used by the handler to parse JSON string (or dict) parameters into a normalized dict for YDB execution.
def _parse_params_str(params_str: str | dict) -> dict: """Parse a JSON params string (or dict) and normalize it for YDB.""" if not params_str: return {} if isinstance(params_str, dict): return _build_ydb_params(params_str) if not params_str.strip(): return {} return _build_ydb_params(json.loads(params_str)) - ydb_mcp/tools.py:100-111 (registration)Registration of the ydb_query_with_params handler with the server via server.add_tool(), mapping YDBGenericTool.QUERY_WITH_PARAMS to the async function with its description.
for tool, fn, description in [ (YDBGenericTool.QUERY, ydb_query, "Run a SQL query against YDB database"), (YDBGenericTool.QUERY_WITH_PARAMS, ydb_query_with_params, "Run a parameterized SQL query with JSON parameters"), (YDBGenericTool.EXPLAIN, ydb_explain_query, "Explain a SQL query against YDB"), (YDBGenericTool.EXPLAIN_WITH_PARAMS, ydb_explain_query_with_params, "Explain a parameterized SQL query against YDB"), (YDBGenericTool.STATUS, ydb_status, "Get the current YDB connection status"), (YDBGenericTool.LIST_DIRECTORY, ydb_list_directory, "List directory contents in YDB"), (YDBGenericTool.DESCRIBE_PATH, ydb_describe_path, "Get detailed information about a YDB path"), ]: if tool in enabled: server.add_tool(fn, name=tool.value, description=description) # type: ignore[arg-type] - ydb_mcp/server.py:115-115 (registration)The call to register_generic_tools(self, type(self).generic_tools) in YDBMCPServer.__init__ which triggers registration of all enabled generic tools including ydb_query_with_params.
register_generic_tools(self, type(self).generic_tools)