list_objects

Instructions

Implementation Reference

• src/postgres_mcp/server.py:124-188 (handler)

  The 'list_objects' tool handler function. Accepts schema_name and object_type (table, view, sequence, or extension) and queries the database via SqlDriver/SafeSqlDriver returning formatted text results.

  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(objects)
      except Exception as e:
          logger.error(f"Error listing objects: {e}")
          return format_error_response(str(e))

• src/postgres_mcp/server.py:124-126 (schema)

  Input schema for list_objects defined via type annotations and Pydantic Field. Parameters: schema_name (str), object_type (str, default='table'). Output is ResponseType (list of TextContent/ImageContent/EmbeddedResource).

  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"),

• src/postgres_mcp/server.py:117-123 (registration)

  Tool registration via the @mcp.tool decorator on the FastMCP 'mcp' instance, with description and ToolAnnotations metadata including readOnlyHint=True.

  @mcp.tool(
      description="List objects in a schema",
      annotations=ToolAnnotations(
          title="List Objects",
          readOnlyHint=True,
      ),
  )