postgres-mcp

Instructions

Implementation Reference

• postgres_mcp/server.py:165-196 (handler)

  The primary MCP tool handler for 'describe_table', registered via @mcp.tool() decorator. It invokes the PostgresClient method and formats the response using ColumnInfo models.
  @mcp.tool() @handle_db_error def describe_table(table_name: str, schema: str = "public") -> dict: """Describe the structure of a table including columns, types, and constraints. Args: table_name: Name of the table to describe schema: Schema name (default: public) Returns: Table structure with columns, primary keys, and foreign keys """ client = get_client() result = client.describe_table(table_name, schema) if not result["columns"]: return not_found_response("Table", f"{schema}.{table_name}") # Transform columns columns = [ ColumnInfo.from_row(col, result["primary_keys"]).model_dump() for col in result["columns"] ] return { "schema": schema, "table_name": table_name, "columns": columns, "primary_keys": result["primary_keys"], "foreign_keys": result["foreign_keys"], }
• postgres_mcp/postgres_client.py:222-295 (helper)

  Core implementation in PostgresClient class that executes SQL queries against information_schema to retrieve table structure details: columns, PKs, and FKs.
  def describe_table(self, table_name: str, schema: str = "public") -> dict[str, Any]: """Get detailed table information. Args: table_name: Table name schema: Schema name (default: public) Returns: Dict with columns, primary keys, foreign keys """ result = { "schema": schema, "name": table_name, "columns": [], "primary_keys": [], "foreign_keys": [], } with self.get_cursor() as cursor: # Get columns cursor.execute(""" SELECT column_name, data_type, is_nullable, column_default, character_maximum_length, numeric_precision, numeric_scale FROM information_schema.columns WHERE table_schema = %s AND table_name = %s ORDER BY ordinal_position """, (schema, table_name)) result["columns"] = [dict(row) for row in cursor.fetchall()] # Get primary keys cursor.execute(""" SELECT column_name FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name AND tc.table_schema = kcu.table_schema WHERE tc.table_schema = %s AND tc.table_name = %s AND tc.constraint_type = 'PRIMARY KEY' """, (schema, table_name)) result["primary_keys"] = [row["column_name"] for row in cursor.fetchall()] # Get foreign keys cursor.execute(""" SELECT kcu.column_name, ccu.table_name AS foreign_table_name, ccu.column_name AS foreign_column_name FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name AND tc.table_schema = kcu.table_schema JOIN information_schema.constraint_column_usage ccu ON ccu.constraint_name = tc.constraint_name AND ccu.table_schema = tc.table_schema WHERE tc.table_schema = %s AND tc.table_name = %s AND tc.constraint_type = 'FOREIGN KEY' """, (schema, table_name)) result["foreign_keys"] = [ { "column": row["column_name"], "references": f"{row['foreign_table_name']}.{row['foreign_column_name']}" } for row in cursor.fetchall() ] return result
• postgres_mcp/models.py:52-77 (schema)

  Pydantic model ColumnInfo used in the handler to validate and structure the list of table columns in the tool response.
  class ColumnInfo(BaseModel): """Column information.""" name: str type: str nullable: bool = True default: Optional[str] = None is_primary_key: bool = False max_length: Optional[int] = None precision: Optional[int] = None scale: Optional[int] = None @classmethod def from_row(cls, row: dict, primary_keys: list[str] = None) -> "ColumnInfo": primary_keys = primary_keys or [] return cls( name=row.get("column_name", ""), type=row.get("data_type", ""), nullable=row.get("is_nullable", "YES") == "YES", default=row.get("column_default"), is_primary_key=row.get("column_name", "") in primary_keys, max_length=row.get("character_maximum_length"), precision=row.get("numeric_precision"), scale=row.get("numeric_scale"), )