YDB MCP

Instructions

Implementation Reference

• ydb_mcp/server.py:704-767 (handler)

  The main handler function for the ydb_list_directory tool. It uses the YDB scheme_client.list_directory(path) to retrieve directory contents, processes entries using ENTRY_TYPE_MAP, handles permissions, sorts items, and returns formatted JSON response.

  async def list_directory(self, path: str) -> List[TextContent]:
      """List the contents of a YDB directory.
  
      Args:
          path: Path to the directory to list
  
      Returns:
          List of TextContent objects with JSON-formatted directory contents
      """
      # Check for authentication errors
      if self.auth_error:
          return [TextContent(type="text", text=json.dumps({"error": self.auth_error}, indent=2))]
  
      try:
          # Create driver if needed
          if self.driver is None:
              await self.create_driver()
  
          if self.driver is None:
              return [TextContent(type="text", text=json.dumps({"error": "Failed to create driver"}, indent=2))]
  
          # Access the scheme client
          scheme_client = self.driver.scheme_client
  
          # List the directory
          logger.info(f"Listing directory contents for path: {path}")
          dir_response = await scheme_client.list_directory(path)
  
          # Process the response
          result = {"path": path, "items": []}
  
          if dir_response.children:
              for entry in dir_response.children:
                  item = {
                      "name": entry.name,
                      "type": self.ENTRY_TYPE_MAP.get(entry.type, str(entry.type)),
                      "owner": entry.owner,
                  }
  
                  # Add permissions if available
                  if hasattr(entry, "permissions") and entry.permissions:
                      item["permissions"] = []
                      for perm in entry.permissions:
                          item["permissions"].append(
                              {
                                  "subject": perm.subject,
                                  "permission_names": list(perm.permission_names),
                              }
                          )
  
                  result["items"].append(item)
  
              # Sort items by name for consistency
              result["items"].sort(key=lambda x: x["name"])
  
          # Convert all dict keys to strings for JSON serialization
          safe_result = self._stringify_dict_keys(result)
          return [TextContent(type="text", text=json.dumps(safe_result, indent=2, cls=CustomJSONEncoder))]
  
      except Exception as e:
          logger.exception(f"Error listing directory {path}: {e}")
          safe_error = self._stringify_dict_keys({"error": f"Error listing directory {path}: {str(e)}"})
          return [TextContent(type="text", text=json.dumps(safe_error, indent=2))]

• ydb_mcp/server.py:618-627 (registration)

  The tool specification dictionary in register_tools method, defining the name, description, handler reference, and parameters schema for registration with FastMCP and ToolManager.

  {
      "name": "ydb_list_directory",
      "description": "List directory contents in YDB",
      "handler": self.list_directory,
      "parameters": {
          "properties": {"path": {"type": "string", "title": "Path"}},
          "required": ["path"],
          "type": "object",
      },
  },

• ydb_mcp/server.py:1066-1067 (registration)

  Specific dispatching logic in the call_tool method that handles calls to ydb_list_directory by invoking the list_directory handler with the path parameter.

  elif tool_name == "ydb_list_directory" and "path" in params:
      result = await self.list_directory(path=params["path"])

• ydb_mcp/server.py:79-92 (helper)

  Class constant mapping YDB entry type integers to human-readable strings, used in the list_directory handler to label directory items.

  ENTRY_TYPE_MAP = {
      1: "DIRECTORY",
      2: "TABLE",
      3: "PERS_QUEUE",
      4: "DATABASE",
      5: "RTMR_VOLUME",
      6: "BLOCK_STORE_VOLUME",
      7: "COORDINATION",
      8: "SEQUENCE",
      9: "REPLICATION",
      10: "TOPIC",
      11: "EXTERNAL_DATA_SOURCE",
      12: "EXTERNAL_TABLE",
  }