OCI MCP Server

Instructions

List DB Nodes in a compartment, optionally filtered by DB System.
Note: compartment_id is always required by the SDK.

Implementation Reference

• mcp_server_oci/tools/dbsystems.py:72-127 (handler)

  Core implementation of list_db_nodes tool handler. Calls OCI DatabaseClient.list_db_nodes API with pagination, extracts and formats node details (id, db_system_id, hostname, vnic_id, lifecycle_state, storage size, time_created). Supports filtering by db_system_id or lists from all systems in compartment.

  def list_db_nodes(
      database_client: oci.database.DatabaseClient,
      db_system_id: Optional[str] = None,
      compartment_id: Optional[str] = None
  ) -> List[Dict[str, Any]]:
      """
      List DB Nodes for a DB System, or for all DB Systems in a compartment.
      Always requires compartment_id for the SDK call.
      """
      try:
          if not compartment_id:
              raise ValueError("compartment_id is required")
  
          nodes: List[Dict[str, Any]] = []
  
          if db_system_id:
              # Correct usage: positional compartment_id + snake_case db_system_id
              resp = oci.pagination.list_call_get_all_results(
                  database_client.list_db_nodes,
                  compartment_id,
                  db_system_id=db_system_id,
              )
              for n in resp.data:
                  nodes.append({
                      "id": n.id,
                      "db_system_id": n.db_system_id,
                      "hostname": getattr(n, "hostname", None),
                      "vnic_id": getattr(n, "vnic_id", None),
                      "lifecycle_state": n.lifecycle_state,
                      "software_storage_size_in_gb": getattr(n, "software_storage_size_in_gb", None),
                      "time_created": str(getattr(n, "time_created", "")),
                  })
          else:
              systems = list_db_systems(database_client, compartment_id)
              for sys in systems:
                  resp = oci.pagination.list_call_get_all_results(
                      database_client.list_db_nodes,
                      compartment_id,
                      db_system_id=sys["id"],
                  )
                  for n in resp.data:
                      nodes.append({
                          "id": n.id,
                          "db_system_id": n.db_system_id,
                          "hostname": getattr(n, "hostname", None),
                          "vnic_id": getattr(n, "vnic_id", None),
                          "lifecycle_state": n.lifecycle_state,
                          "software_storage_size_in_gb": getattr(n, "software_storage_size_in_gb", None),
                          "time_created": str(getattr(n, "time_created", "")),
                      })
  
          logger.info(f"Found {len(nodes)} DB Nodes")
          return nodes
      except Exception as e:
          logger.exception(f"Error listing DB Nodes: {e}")
          raise

• mcp_server_oci/mcp_server.py:496-510 (registration)

  MCP tool registration for "list_db_nodes". Uses @mcp.tool decorator on a wrapper function that injects oci_clients["database"] client and calls the core handler from dbsystems.py. Input schema inferred from signature: compartment_id (str, required), db_system_id (str, optional).

  @mcp.tool(name="list_db_nodes")
  @mcp_tool_wrapper(
      start_msg="Listing DB Nodes in compartment {compartment_id}...",
      error_prefix="Error listing DB Nodes"
  )
  async def mcp_list_db_nodes(ctx: Context, compartment_id: str, db_system_id: Optional[str] = None) -> List[Dict[str, Any]]:
      """
      List DB Nodes in a compartment, optionally filtered by DB System.
      Note: compartment_id is always required by the SDK.
      """
      return list_db_nodes(
          oci_clients["database"],
          db_system_id=db_system_id,
          compartment_id=compartment_id,
      )