Postgres MCP

Instructions

Implementation Reference

• src/postgres_mcp/server.py:177-307 (handler)

  The handler function for the 'get_object_details' tool. It retrieves detailed information about database objects (tables, views, sequences, extensions) including columns, data types, nullability, defaults, constraints, and indexes by executing appropriate SQL queries against information_schema and pg_catalog.

  @mcp.tool(description="Show detailed information about a database object")
  async def get_object_details(
      schema_name: str = Field(description="Schema name"),
      object_name: str = Field(description="Object name"),
      object_type: str = Field(description="Object type: 'table', 'view', 'sequence', or 'extension'", default="table"),
  ) -> ResponseType:
      """Get detailed information about a database object."""
      try:
          sql_driver = await get_sql_driver()
  
          if object_type in ("table", "view"):
              # Get columns
              col_rows = await SafeSqlDriver.execute_param_query(
                  sql_driver,
                  """
                  SELECT column_name, data_type, is_nullable, column_default
                  FROM information_schema.columns
                  WHERE table_schema = {} AND table_name = {}
                  ORDER BY ordinal_position
                  """,
                  [schema_name, object_name],
              )
              columns = (
                  [
                      {
                          "column": r.cells["column_name"],
                          "data_type": r.cells["data_type"],
                          "is_nullable": r.cells["is_nullable"],
                          "default": r.cells["column_default"],
                      }
                      for r in col_rows
                  ]
                  if col_rows
                  else []
              )
  
              # Get constraints
              con_rows = await SafeSqlDriver.execute_param_query(
                  sql_driver,
                  """
                  SELECT tc.constraint_name, tc.constraint_type, kcu.column_name
                  FROM information_schema.table_constraints AS tc
                  LEFT JOIN information_schema.key_column_usage AS kcu
                    ON tc.constraint_name = kcu.constraint_name
                   AND tc.table_schema = kcu.table_schema
                  WHERE tc.table_schema = {} AND tc.table_name = {}
                  """,
                  [schema_name, object_name],
              )
  
              constraints = {}
              if con_rows:
                  for row in con_rows:
                      cname = row.cells["constraint_name"]
                      ctype = row.cells["constraint_type"]
                      col = row.cells["column_name"]
  
                      if cname not in constraints:
                          constraints[cname] = {"type": ctype, "columns": []}
                      if col:
                          constraints[cname]["columns"].append(col)
  
              constraints_list = [{"name": name, **data} for name, data in constraints.items()]
  
              # Get indexes
              idx_rows = await SafeSqlDriver.execute_param_query(
                  sql_driver,
                  """
                  SELECT indexname, indexdef
                  FROM pg_indexes
                  WHERE schemaname = {} AND tablename = {}
                  """,
                  [schema_name, object_name],
              )
  
              indexes = [{"name": r.cells["indexname"], "definition": r.cells["indexdef"]} for r in idx_rows] if idx_rows else []
  
              result = {
                  "basic": {"schema": schema_name, "name": object_name, "type": object_type},
                  "columns": columns,
                  "constraints": constraints_list,
                  "indexes": indexes,
              }
  
          elif object_type == "sequence":
              rows = await SafeSqlDriver.execute_param_query(
                  sql_driver,
                  """
                  SELECT sequence_schema, sequence_name, data_type, start_value, increment
                  FROM information_schema.sequences
                  WHERE sequence_schema = {} AND sequence_name = {}
                  """,
                  [schema_name, object_name],
              )
  
              if rows and rows[0]:
                  row = rows[0]
                  result = {
                      "schema": row.cells["sequence_schema"],
                      "name": row.cells["sequence_name"],
                      "data_type": row.cells["data_type"],
                      "start_value": row.cells["start_value"],
                      "increment": row.cells["increment"],
                  }
              else:
                  result = {}
  
          elif object_type == "extension":
              rows = await SafeSqlDriver.execute_param_query(
                  sql_driver,
                  """
                  SELECT extname, extversion, extrelocatable
                  FROM pg_extension
                  WHERE extname = {}
                  """,
                  [object_name],
              )
  
              if rows and rows[0]:
                  row = rows[0]
                  result = {"name": row.cells["extname"], "version": row.cells["extversion"], "relocatable": row.cells["extrelocatable"]}
              else:
                  result = {}
  
          else:
              return format_error_response(f"Unsupported object type: {object_type}")
  
          return format_text_response(result)
      except Exception as e:
          logger.error(f"Error getting object details: {e}")
          return format_error_response(str(e))