ydb_explain_query_with_params

Instructions

Implementation Reference

• ydb_mcp/tools.py:56-62 (handler)

  The ydb_explain_query_with_params handler function: takes sql and params, parses params via _parse_params_str, calls server.explain(sql, params), serializes the result, and returns TextContent. Wraps in try/except to return error JSON on failure.

  async def ydb_explain_query_with_params(sql: str, params: str | dict) -> list[TextContent]:
      """Explain a parameterized SQL query against YDB."""
      try:
          result = await server.explain(sql, _parse_params_str(params))
          return [TextContent(type="text", text=serialize_ydb_response(result))]
      except Exception as e:
          return [TextContent(type="text", text=json.dumps({"error": str(e)}, indent=2))]

• ydb_mcp/tools.py:16-25 (schema)

  The YDBGenericTool enum defines EXPLAIN_WITH_PARAMS = 'ydb_explain_query_with_params' as the tool name string.

  class YDBGenericTool(str, Enum):
      """Names of the built-in generic YDB tools."""
  
      QUERY = "ydb_query"
      QUERY_WITH_PARAMS = "ydb_query_with_params"
      EXPLAIN = "ydb_explain_query"
      EXPLAIN_WITH_PARAMS = "ydb_explain_query_with_params"
      STATUS = "ydb_status"
      LIST_DIRECTORY = "ydb_list_directory"
      DESCRIBE_PATH = "ydb_describe_path"

• ydb_mcp/tools.py:100-111 (registration)

  The registration loop in register_generic_tools maps YDBGenericTool.EXPLAIN_WITH_PARAMS to the ydb_explain_query_with_params handler with description, then calls server.add_tool() if the tool is in the enabled set.

  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:171-181 (helper)

  The server.explain() method called by the handler: ensures connection, builds YDB params via _build_ydb_params, calls self._pool.explain_with_retries() with query, parameters, and result_format=ydb.QueryExplainResultFormat.DICT, then stringifies keys for JSON.

  async def explain(self, sql: str, params: dict | None = None) -> dict:
      """Explain a SQL query and return the execution plan as a dict."""
      await self._ensure_connected()
      assert self._pool is not None
      ydb_params = _build_ydb_params(params) if params else None
      plan = await self._pool.explain_with_retries(
          query=sql,
          parameters=ydb_params,
          result_format=ydb.QueryExplainResultFormat.DICT,
      )
      return dict(_stringify_keys(plan))

• ydb_mcp/params.py:24-32 (helper)

  The _parse_params_str helper used by the handler to parse parameters from JSON string or dict into normalized YDB params.

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