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