Postgres MCP

Instructions

Implementation Reference

• src/postgres_mcp/server.py:109-175 (handler)

  The main handler function for the 'list_objects' tool. It lists database objects (tables, views, sequences, or extensions) in a specified schema by executing appropriate SQL queries against information_schema and pg_extension. The function is decorated with @mcp.tool, which also serves as registration. Input schema is defined inline using Pydantic Field.

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