YDB MCP

Instructions

Implementation Reference

• ydb_mcp/server.py:768-955 (handler)

  The primary handler function for the 'ydb_describe_path' tool. It uses YDB's scheme_client.describe_path to retrieve basic path information and, if the path is a table, uses table_client to fetch detailed table schema including columns, indexes, partitioning settings, etc. Returns formatted JSON description.
  async def describe_path(self, path: str) -> List[TextContent]: """Describe a path in YDB. Args: path: Path to describe Returns: List of TextContent objects with path description """ # Check for authentication errors if self.auth_error: safe_error = {"error": self.auth_error} return [TextContent(type="text", text=json.dumps(safe_error, indent=2))] try: # Create driver if needed if self.driver is None: await self.create_driver() if self.driver is None: safe_error = {"error": "Failed to create driver"} return [TextContent(type="text", text=json.dumps(safe_error, indent=2))] # Access the scheme client scheme_client = self.driver.scheme_client # Describe the path logger.info(f"Describing path: {path}") path_response = await scheme_client.describe_path(path) # Process the response if path_response is None: safe_error = {"error": f"Path '{path}' not found"} return [TextContent(type="text", text=json.dumps(safe_error, indent=2))] # Format the result result = { "path": path, "type": str(path_response.type), "name": path_response.name, "owner": path_response.owner, } # Add permissions if available if hasattr(path_response, "permissions") and path_response.permissions: result["permissions"] = [] for perm in path_response.permissions: result["permissions"].append( {"subject": perm.subject, "permission_names": list(perm.permission_names)} ) # Add table specific information if it's a table if str(path_response.type) == "TABLE" or path_response.type == 2: try: # Get table client for more detailed table info table_client = self.driver.table_client session = await table_client.session().create() try: # Get detailed table description table_desc = await session.describe_table(path) result["table"] = { "columns": [], "primary_key": table_desc.primary_key, "indexes": [], "partitioning_settings": {}, "storage_settings": {}, "key_bloom_filter": table_desc.key_bloom_filter, "read_replicas_settings": table_desc.read_replicas_settings, "column_families": [], } # Add columns with more details for column in table_desc.columns: col_info = { "name": column.name, "type": str(column.type), "family": column.family, } result["table"]["columns"].append(col_info) # Add indexes with more details for index in table_desc.indexes: index_info = { "name": index.name, "index_columns": list(index.index_columns), "cover_columns": (list(index.cover_columns) if hasattr(index, "cover_columns") else []), "index_type": (str(index.index_type) if hasattr(index, "index_type") else None), } result["table"]["indexes"].append(index_info) # Add column families if present if hasattr(table_desc, "column_families"): for family in table_desc.column_families: family_info = { "name": family.name, "data": family.data, "compression": (str(family.compression) if hasattr(family, "compression") else None), } result["table"]["column_families"].append(family_info) # Add storage settings if present if hasattr(table_desc, "storage_settings"): ss = table_desc.storage_settings if ss: result["table"]["storage_settings"] = { "tablet_commit_log0": ss.tablet_commit_log0, "tablet_commit_log1": ss.tablet_commit_log1, "external": ss.external, "store_external": ss.store_external, } # Add partitioning settings if present if hasattr(table_desc, "partitioning_settings"): ps = table_desc.partitioning_settings if ps: if hasattr(ps, "partition_at_keys"): result["table"]["partitioning_settings"]["partition_at_keys"] = ps.partition_at_keys if hasattr(ps, "partition_by_size"): result["table"]["partitioning_settings"]["partition_by_size"] = ps.partition_by_size if hasattr(ps, "min_partitions_count"): result["table"]["partitioning_settings"]["min_partitions_count"] = ( ps.min_partitions_count ) if hasattr(ps, "max_partitions_count"): result["table"]["partitioning_settings"]["max_partitions_count"] = ( ps.max_partitions_count ) finally: # Always release the session await session.close() except Exception as table_error: logger.warning(f"Error getting detailed table info: {table_error}") # Fallback to basic table info from path_response if hasattr(path_response, "table") and path_response.table: result["table"] = { "columns": [], "primary_key": ( path_response.table.primary_key if hasattr(path_response.table, "primary_key") else [] ), "indexes": [], "partitioning_settings": {}, } # Add basic columns if hasattr(path_response.table, "columns"): for column in path_response.table.columns: result["table"]["columns"].append({"name": column.name, "type": str(column.type)}) # Add basic indexes if hasattr(path_response.table, "indexes"): for index in path_response.table.indexes: result["table"]["indexes"].append( { "name": index.name, "index_columns": ( list(index.index_columns) if hasattr(index, "index_columns") else [] ), } ) # Add basic partitioning settings if hasattr(path_response.table, "partitioning_settings"): ps = path_response.table.partitioning_settings if ps: if hasattr(ps, "partition_at_keys"): result["table"]["partitioning_settings"]["partition_at_keys"] = ps.partition_at_keys if hasattr(ps, "partition_by_size"): result["table"]["partitioning_settings"]["partition_by_size"] = ps.partition_by_size if hasattr(ps, "min_partitions_count"): result["table"]["partitioning_settings"]["min_partitions_count"] = ( ps.min_partitions_count ) if hasattr(ps, "max_partitions_count"): result["table"]["partitioning_settings"]["max_partitions_count"] = ( ps.max_partitions_count ) # Convert to JSON string and return as TextContent formatted_result = json.dumps(result, indent=2, cls=CustomJSONEncoder) return [TextContent(type="text", text=formatted_result)] except Exception as e: logger.exception(f"Error describing path {path}: {e}") safe_error = {"error": f"Error describing path {path}: {str(e)}"} return [TextContent(type="text", text=json.dumps(safe_error, indent=2))]
• ydb_mcp/server.py:628-658 (registration)

  Tool registration in register_tools method. Defines the tool specification including name, description, handler reference, and input schema (parameters). Registers the tool with both FastMCP (self.add_tool) and the internal ToolManager.
  { "name": "ydb_describe_path", "description": "Get detailed information about a YDB path (table, directory, etc.)", "handler": self.describe_path, "parameters": { "properties": {"path": {"type": "string", "title": "Path"}}, "required": ["path"], "type": "object", }, }, ] # Register all tools with FastMCP framework 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:628-637 (schema)

  JSON schema definition for the tool's input parameters: requires a 'path' string.
  { "name": "ydb_describe_path", "description": "Get detailed information about a YDB path (table, directory, etc.)", "handler": self.describe_path, "parameters": { "properties": {"path": {"type": "string", "title": "Path"}}, "required": ["path"], "type": "object", }, },
• ydb_mcp/server.py:1068-1069 (handler)

  Dispatch logic in the overridden call_tool method that routes calls to the describe_path handler.
  elif tool_name == "ydb_describe_path" and "path" in params: result = await self.describe_path(path=params["path"])