Understanding MCP(Model Context Protocol) Servers: The Future of AI Communication

# Resource: Get table schema
@mcp.resource("postgres://schemas/{schema}/tables/{table}/schema")
def get_table_schema(schema: str, table: str) -> str:
    """Get the schema for a specific table"""
    try:
        if not re.match(r'^[a-zA-Z0-9_]+$', schema) or not re.match(r'^[a-zA-Z0-9_]+$', table):
            return "Invalid schema or table name"
            
        # Get columns and primary keys
        columns_result = safe_query("""
            SELECT column_name, data_type, is_nullable, column_default
            FROM information_schema.columns
            WHERE table_schema = %s AND table_name = %s
            ORDER BY ordinal_position
        """, [schema, table])
        
        pk_result = safe_query("""
            SELECT c.column_name
            FROM information_schema.table_constraints tc
            JOIN information_schema.constraint_column_usage AS ccu USING (constraint_schema, constraint_name)
            JOIN information_schema.columns AS c 
              ON c.table_schema = tc.constraint_schema
              AND c.table_name = tc.table_name
              AND c.column_name = ccu.column_name
            WHERE tc.constraint_type = 'PRIMARY KEY' AND tc.table_schema = %s AND tc.table_name = %s
        """, [schema, table])
        
        primary_keys = [row[0] for row in pk_result["rows"]]
        
        # Format output
        column_lines = []
        for col in columns_result["rows"]:
            col_name, data_type, is_nullable, col_default = col
            is_pk = col_name in primary_keys
            pk_marker = " (PK)" if is_pk else ""
            nullable_marker = "" if is_nullable == 'YES' else " NOT NULL"
            default_marker = f" DEFAULT {col_default}" if col_default else ""
            column_lines.append(f"- {col_name}: {data_type}{pk_marker}{nullable_marker}{default_marker}")
        
        return f"Schema for {schema}.{table}:\n\n" + "\n".join(column_lines)
    except Exception as e:
        return f"Error retrieving schema: {str(e)}"

# Resource: Get table sample
@mcp.resource("postgres://schemas/{schema}/tables/{table}/sample")
def get_table_sample(schema: str, table: str) -> str:
    """Get a sample of data from a table"""
    try:
        if not re.match(r'^[a-zA-Z0-9_]+$', schema) or not re.match(r'^[a-zA-Z0-9_]+$', table):
            return "Invalid schema or table name"
            
        result = safe_query(f'SELECT * FROM "{schema}"."{table}" LIMIT 5')
        return f"Sample data from {schema}.{table}:\n\n{format_table_result(result)}"
    except Exception as e:
        return f"Error retrieving sample: {str(e)}"