YDB MCP

Instructions

Implementation Reference

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

  The core handler function `list_directory` that implements the `ydb_list_directory` tool. It lists the contents of a YDB directory using `scheme_client.list_directory(path)`, processes the entries (name, type via ENTRY_TYPE_MAP, owner, permissions), sorts by name, and returns JSON-formatted results as TextContent.
  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:641-658 (registration)

  The registration loop in `register_tools()` that adds the `ydb_list_directory` tool (along with others) to FastMCP via `add_tool` and to the internal `tool_manager` via `register_tool`, linking to the handler `self.list_directory`.
  for spec in tool_specs: self.add_tool( spec["handler"], name=spec["name"], description=spec["description"], # Structured output is temporarily disabled until proper schema definitions are implemented. # See https://github.com/ydb-platform/ydb-mcp/issues/12 for details. structured_output=False, ) # Also register with our tool manager self.tool_manager.register_tool( name=spec["name"], handler=spec["handler"], description=spec["description"], parameters=spec.get("parameters"), )
• ydb_mcp/server.py:619-627 (schema)

  Tool specification in `tool_specs` list including the JSON schema for input parameters: requires a `path` string.
  "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:79-92 (helper)

  Static mapping from YDB entry type integers to human-readable strings, used in the `list_directory` handler to label directory items (e.g., 2 -> "TABLE").
  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", }
• ydb_mcp/server.py:1066-1067 (handler)

  Dispatch logic in `call_tool` method that routes calls to `ydb_list_directory` to the `list_directory` handler with extracted `path` parameter.
  elif tool_name == "ydb_list_directory" and "path" in params: result = await self.list_directory(path=params["path"])