YDB MCP

Instructions

Implementation Reference

• ydb_mcp/server.py:504-532 (handler)

  Primary handler function that implements the core logic for the 'ydb_query_with_params' tool. Parses the JSON params string and executes the query using the shared 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 specification in the register_tools method, defining name, description, handler reference, and parameters schema. This spec is used to register the tool with both FastMCP.add_tool and ToolManager.register_tool.

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

  JSON schema defining the input parameters for the ydb_query_with_params tool: sql (string) and params (string). Used for validation.

  "parameters": {
      "properties": {
          "sql": {"type": "string", "title": "Sql"},
          "params": {"type": "string", "title": "Params"},
      },
      "required": ["sql", "params"],
      "type": "object",
  },

• ydb_mcp/server.py:533-553 (helper)

  Supporting helper function called by the handler to parse the input params JSON string into a YDB-compatible parameters 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 code in the call_tool method that specifically routes requests for this tool to the query_with_params 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"])