list_objects

Instructions

Implementation Reference

• src/postgres_mcp_pro_plus/server.py:240-327 (handler)

  The handler function implementing the 'list_objects' tool. It queries the database for objects of the specified type in the given schema (or all extensions), formats the results, and returns them as text. Includes input schema via Pydantic Fields and registration via @mcp.tool decorator.

  @mcp.tool(description="List objects in a schema")
  async def list_objects(
      schema_name: str = Field(description="Schema name"),
      object_type: str = Field(description="Object type: 'table', 'view', 'sequence', or 'extension'", default="table"),
  ) -> ResponseType:
      """List objects of a given type in a schema."""
      try:
          sql_driver = await get_sql_driver()
  
          if object_type in ("table", "view"):
              table_type = "BASE TABLE" if object_type == "table" else "VIEW"
              rows = await SafeSqlDriver.execute_param_query(
                  sql_driver,
                  """
                  SELECT table_schema, table_name, table_type
                  FROM information_schema.tables
                  WHERE table_schema = {} AND table_type = {}
                  ORDER BY table_name
                  """,
                  [schema_name, table_type],
              )
              objects = (
                  [
                      {
                          "schema": row.cells["table_schema"],
                          "name": row.cells["table_name"],
                          "type": row.cells["table_type"],
                      }
                      for row in rows
                  ]
                  if rows
                  else []
              )
  
          elif object_type == "sequence":
              rows = await SafeSqlDriver.execute_param_query(
                  sql_driver,
                  """
                  SELECT sequence_schema, sequence_name, data_type
                  FROM information_schema.sequences
                  WHERE sequence_schema = {}
                  ORDER BY sequence_name
                  """,
                  [schema_name],
              )
              objects = (
                  [
                      {
                          "schema": row.cells["sequence_schema"],
                          "name": row.cells["sequence_name"],
                          "data_type": row.cells["data_type"],
                      }
                      for row in rows
                  ]
                  if rows
                  else []
              )
  
          elif object_type == "extension":
              # Extensions are not schema-specific
              rows = await sql_driver.execute_query(
                  """
                  SELECT extname, extversion, extrelocatable
                  FROM pg_extension
                  ORDER BY extname
                  """
              )
              objects = (
                  [
                      {
                          "name": row.cells["extname"],
                          "version": row.cells["extversion"],
                          "relocatable": row.cells["extrelocatable"],
                      }
                      for row in rows
                  ]
                  if rows
                  else []
              )
  
          else:
              return format_error_response(f"Unsupported object type: {object_type}")
  
          return format_text_response(format_objects_as_text(objects, object_type))
      except Exception as e:
          logger.error(f"Error listing objects: {e}")
          return format_error_response(str(e))

• src/postgres_mcp_pro_plus/server.py:106-125 (helper)

  Helper function to format the list of database objects into a compact text string used by the list_objects tool.

  def format_objects_as_text(objects: list[dict], object_type: str) -> str:
      """Format object lists compactly without emojis/headers, preserving details."""
      if not objects:
          return f"No {object_type}s found."
  
      label_map = {"table": "Tables", "view": "Views", "sequence": "Sequences", "extension": "Extensions"}
      label = label_map.get(object_type, object_type.capitalize() + "s")
  
      def item_str(o: dict) -> str:
          if object_type in ("table", "view"):
              return f"{o['schema']}.{o['name']}({o['type']})"
          if object_type == "sequence":
              return f"{o['schema']}.{o['name']}({o['data_type']})"
          if object_type == "extension":
              return f"{o['name']} v{o['version']} reloc={o['relocatable']}"
          return str(o)
  
      items = "; ".join(item_str(o) for o in objects)
      return f"{label}({len(objects)}): {items}"