# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""GlueCommonsHandler for Data Processing MCP Server."""

from awslabs.aws_dataprocessing_mcp_server.models.glue_models import (
    CreateSecurityConfigurationData,
    CreateUsageProfileData,
    DeleteResourcePolicyData,
    DeleteSecurityConfigurationData,
    DeleteUsageProfileData,
    GetDataCatalogEncryptionSettingsData,
    GetResourcePolicyData,
    GetSecurityConfigurationData,
    GetUsageProfileData,
    PutDataCatalogEncryptionSettingsData,
    PutResourcePolicyData,
    UpdateUsageProfileData,
)
from awslabs.aws_dataprocessing_mcp_server.utils.aws_helper import AwsHelper
from awslabs.aws_dataprocessing_mcp_server.utils.logging_helper import (
    LogLevel,
    log_with_request_id,
)
from botocore.exceptions import ClientError
from mcp.server.fastmcp import Context
from mcp.types import CallToolResult, TextContent
from pydantic import Field
from typing import Annotated, Any, Dict, Optional


class GlueCommonsHandler:
    """Handler for Amazon Glue common operations."""

    def __init__(self, mcp, allow_write: bool = False, allow_sensitive_data_access: bool = False):
        """Initialize the Glue Commons handler.

        Args:
            mcp: The MCP server instance
            allow_write: Whether to enable write access (default: False)
            allow_sensitive_data_access: Whether to allow access to sensitive data (default: False)
        """
        self.mcp = mcp
        self.allow_write = allow_write
        self.allow_sensitive_data_access = allow_sensitive_data_access
        self.glue_client = AwsHelper.create_boto3_client('glue')

        # Register tools
        self.mcp.tool(name='manage_aws_glue_usage_profiles')(self.manage_aws_glue_usage_profiles)
        self.mcp.tool(name='manage_aws_glue_security_configurations')(
            self.manage_aws_glue_security
        )
        self.mcp.tool(name='manage_aws_glue_encryption')(self.manage_aws_glue_encryption)
        self.mcp.tool(name='manage_aws_glue_resource_policies')(
            self.manage_aws_glue_resource_policies
        )

    async def manage_aws_glue_usage_profiles(
        self,
        ctx: Context,
        operation: Annotated[
            str,
            Field(
                description='Operation to perform: create-profile, delete-profile, get-profile, update-profile. Choose "get-profile" for read-only operations when write access is disabled.',
            ),
        ],
        profile_name: Annotated[
            str,
            Field(
                description='Name of the usage profile.',
            ),
        ],
        description: Annotated[
            Optional[str],
            Field(
                description='Description of the usage profile (for create-profile and update-profile operations).',
            ),
        ] = None,
        configuration: Annotated[
            Optional[Dict[str, Any]],
            Field(
                description='Configuration object specifying job and session values for the profile (required for create-profile and update-profile operations).',
            ),
        ] = None,
        tags: Annotated[
            Optional[Dict[str, str]],
            Field(
                description='Tags to apply to the usage profile (for create-profile operation).',
            ),
        ] = None,
    ) -> CallToolResult:
        """Manage AWS Glue Usage Profiles for resource allocation and cost management.

        This tool allows you to create, retrieve, update, and delete AWS Glue Usage Profiles, which define
        resource allocation and cost management settings for Glue jobs and interactive sessions.

        ## Requirements
        - The server must be run with the `--allow-write` flag for create-profile, delete-profile, and update-profile operations
        - Appropriate AWS permissions for Glue Usage Profile operations

        ## Operations
        - **create-profile**: Create a new usage profile with specified resource allocations
        - **delete-profile**: Delete an existing usage profile
        - **get-profile**: Retrieve detailed information about a specific usage profile
        - **update-profile**: Update an existing usage profile's configuration

        ## Example
        ```json
        {
          "operation": "create-profile",
          "profile_name": "my-standard-profile",
          "description": "Standard resource allocation for ETL jobs",
          "configuration": {
              "JobConfiguration": {
                "numberOfWorkers": {
                  "DefaultValue": "10",
                  "MinValue": "1",
                  "MaxValue": "10"
                },
                "workerType": {
                  "DefaultValue": "G.2X",
                  "AllowedValues": [
                    "G.2X",
                    "G.4X",
                    "G.8X"
                  ]
                },
            }
        }
        ```

        Args:
            ctx: MCP context
            operation: Operation to perform
            profile_name: Name of the usage profile
            description: Description of the usage profile
            configuration: Configuration object specifying job and session values
            tags: Tags to apply to the usage profile

        Returns:
            Union of response types specific to the operation performed
        """
        try:
            if not self.allow_write and operation != 'get-profile':
                error_message = f'Operation {operation} is not allowed without write access'
                log_with_request_id(ctx, LogLevel.ERROR, error_message)

                return CallToolResult(
                    isError=True,
                    content=[TextContent(type='text', text=error_message)],
                )

            if operation == 'create-profile':
                if configuration is None:
                    raise ValueError('configuration is required for create-profile operation')

                # Prepare create request parameters
                params = {'Name': profile_name, 'Configuration': configuration}

                if description:
                    params['Description'] = description

                # Add MCP management tags
                resource_tags = AwsHelper.prepare_resource_tags('GlueUsageProfile')

                # Merge user-provided tags with MCP tags
                if tags:
                    merged_tags = tags.copy()
                    merged_tags.update(resource_tags)
                    params['Tags'] = merged_tags
                else:
                    params['Tags'] = resource_tags

                # Create the usage profile
                response = self.glue_client.create_usage_profile(**params)

                success_message = f'Successfully created usage profile {profile_name}'
                data = CreateUsageProfileData(
                    profile_name=profile_name,
                    operation='create-usage-profile',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            elif operation == 'delete-profile':
                # First get the profile to check if it's managed by MCP
                try:
                    response = self.glue_client.get_usage_profile(Name=profile_name)

                    # Construct the ARN for the usage profile
                    region = AwsHelper.get_or_default_aws_region() or 'us-east-1'
                    account_id = AwsHelper.get_aws_account_id()
                    profile_arn = f'arn:aws:glue:{region}:{account_id}:usageProfile/{profile_name}'

                    # Check if the profile is managed by MCP
                    tags = response.get('Tags', {})
                    if not AwsHelper.is_resource_mcp_managed(self.glue_client, profile_arn, {}):
                        error_message = f'Cannot delete usage profile {profile_name} - it is not managed by the MCP server (missing required tags)'
                        log_with_request_id(ctx, LogLevel.ERROR, error_message)
                        return CallToolResult(
                            isError=True,
                            content=[TextContent(type='text', text=error_message)],
                        )
                except ClientError as e:
                    if e.response['Error']['Code'] == 'EntityNotFoundException':
                        error_message = f'Usage profile {profile_name} not found'
                        log_with_request_id(ctx, LogLevel.ERROR, error_message)
                        return CallToolResult(
                            isError=True,
                            content=[TextContent(type='text', text=error_message)],
                        )
                    else:
                        raise e

                # Delete the usage profile if it's managed by MCP
                self.glue_client.delete_usage_profile(Name=profile_name)

                success_message = f'Successfully deleted usage profile {profile_name}'
                data = DeleteUsageProfileData(
                    profile_name=profile_name,
                    operation='delete-usage-profile',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            elif operation == 'get-profile':
                # Get the usage profile
                response = self.glue_client.get_usage_profile(Name=profile_name)

                success_message = f'Successfully retrieved usage profile {profile_name}'
                data = GetUsageProfileData(
                    profile_name=response.get('Name', profile_name),
                    profile_details=response,
                    operation='get-usage-profile',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            elif operation == 'update-profile':
                if configuration is None:
                    raise ValueError('configuration is required for update-profile operation')

                # First get the profile to check if it's managed by MCP
                try:
                    response = self.glue_client.get_usage_profile(Name=profile_name)

                    # Construct the ARN for the usage profile
                    region = AwsHelper.get_or_default_aws_region() or 'us-east-1'
                    account_id = AwsHelper.get_aws_account_id()
                    profile_arn = f'arn:aws:glue:{region}:{account_id}:usageProfile/{profile_name}'

                    # Check if the profile is managed by MCP
                    tags = response.get('Tags', {})
                    if not AwsHelper.is_resource_mcp_managed(self.glue_client, profile_arn, {}):
                        error_message = f'Cannot update usage profile {profile_name} - it is not managed by the MCP server (missing required tags)'
                        log_with_request_id(ctx, LogLevel.ERROR, error_message)
                        return CallToolResult(
                            isError=True,
                            content=[TextContent(type='text', text=error_message)],
                        )
                except ClientError as e:
                    if e.response['Error']['Code'] == 'EntityNotFoundException':
                        error_message = f'Usage profile {profile_name} not found'
                        log_with_request_id(ctx, LogLevel.ERROR, error_message)
                        return CallToolResult(
                            isError=True,
                            content=[TextContent(type='text', text=error_message)],
                        )
                    else:
                        raise e

                # Prepare update request parameters
                params = {'Name': profile_name, 'Configuration': configuration}

                if description:
                    params['Description'] = description

                # Update the usage profile
                response = self.glue_client.update_usage_profile(**params)

                success_message = f'Successfully updated usage profile {profile_name}'
                data = UpdateUsageProfileData(
                    profile_name=profile_name,
                    operation='update-usage-profile',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            else:
                error_message = f'Invalid operation: {operation}. Must be one of: create-profile, delete-profile, get-profile, update-profile'
                log_with_request_id(ctx, LogLevel.ERROR, error_message)
                return CallToolResult(
                    isError=True,
                    content=[TextContent(type='text', text=error_message)],
                )

        except ValueError as e:
            log_with_request_id(ctx, LogLevel.ERROR, f'Parameter validation error: {str(e)}')
            raise
        except Exception as e:
            error_message = f'Error in manage_aws_glue_usage_profiles: {str(e)}'
            log_with_request_id(ctx, LogLevel.ERROR, error_message)
            return CallToolResult(
                isError=True,
                content=[TextContent(type='text', text=error_message)],
            )

    async def manage_aws_glue_security(
        self,
        ctx: Context,
        operation: Annotated[
            str,
            Field(
                description='Operation to perform: create-security-configuration, delete-security-configuration, get-security-configuration. Choose "get-security-configuration" for read-only operations when write access is disabled.',
            ),
        ],
        config_name: Annotated[
            str,
            Field(
                description='Name of the security configuration.',
            ),
        ],
        encryption_configuration: Annotated[
            Optional[Dict[str, Any]],
            Field(
                description='Encryption configuration for create-security-configuration operation, containing settings for S3, CloudWatch, and job bookmarks encryption.',
            ),
        ] = None,
    ) -> CallToolResult:
        """Manage AWS Glue Security Configurations for data encryption.

        This tool allows you to create, retrieve, and delete AWS Glue Security Configurations, which define
        encryption settings for Glue jobs, crawlers, and development endpoints.

        ## Requirements
        - The server must be run with the `--allow-write` flag for create-security-configuration and delete-security-configuration operations
        - Appropriate AWS permissions for Glue Security Configuration operations

        ## Operations
        - **create-security-configuration**: Create a new security configuration with encryption settings
        - **delete-security-configuration**: Delete an existing security configuration
        - **get-security-configuration**: Retrieve detailed information about a specific security configuration

        ## Example
        ```json
        {
          "operation": "create-security-configuration",
          "config_name": "my-encryption-config",
          "encryption_configuration": {
            "S3Encryption": [
              {
                "S3EncryptionMode": "SSE-KMS",
                "KmsKeyArn": "arn:aws:kms:region:account-id:key/key-id"
              }
            ],
            "CloudWatchEncryption": {
              "CloudWatchEncryptionMode": "DISABLED"
            },
            "JobBookmarksEncryption": {
              "JobBookmarksEncryptionMode": "CSE-KMS",
              "KmsKeyArn": "arn:aws:kms:region:account-id:key/key-id"
            }
          }
        }
        ```

        Args:
            ctx: MCP context
            operation: Operation to perform
            config_name: Name of the security configuration
            encryption_configuration: Encryption configuration for create-security-configuration operation

        Returns:
            Union of response types specific to the operation performed
        """
        try:
            if not self.allow_write and operation != 'get-security-configuration':
                error_message = f'Operation {operation} is not allowed without write access'
                log_with_request_id(ctx, LogLevel.ERROR, error_message)

                return CallToolResult(
                    isError=True,
                    content=[TextContent(type='text', text=error_message)],
                )

            if operation == 'create-security-configuration':
                if encryption_configuration is None:
                    raise ValueError(
                        'encryption_configuration is required for create-security-configuration operation'
                    )

                # Create the security configuration
                response = self.glue_client.create_security_configuration(
                    Name=config_name, EncryptionConfiguration=encryption_configuration
                )

                success_message = f'Successfully created security configuration {config_name}'
                data = CreateSecurityConfigurationData(
                    config_name=config_name,
                    creation_time=(
                        response.get('CreatedTimestamp', '').isoformat()
                        if response.get('CreatedTimestamp')
                        else ''
                    ),
                    encryption_configuration=encryption_configuration or {},
                    operation='create-security-configuration',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            elif operation == 'delete-security-configuration':
                # First check if the security configuration exists
                try:
                    # Get the security configuration
                    self.glue_client.get_security_configuration(Name=config_name)

                    # Note: Security configurations don't support tags in AWS Glue API
                    # so we can't verify if it's managed by MCP

                except ClientError as e:
                    if e.response['Error']['Code'] == 'EntityNotFoundException':
                        error_message = f'Security configuration {config_name} not found'
                        log_with_request_id(ctx, LogLevel.ERROR, error_message)
                        return CallToolResult(
                            isError=True,
                            content=[TextContent(type='text', text=error_message)],
                        )
                    else:
                        raise e

                # Delete the security configuration
                self.glue_client.delete_security_configuration(Name=config_name)

                success_message = f'Successfully deleted security configuration {config_name}'
                data = DeleteSecurityConfigurationData(
                    config_name=config_name,
                    operation='delete-security-configuration',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            elif operation == 'get-security-configuration':
                # Get the security configuration
                response = self.glue_client.get_security_configuration(Name=config_name)

                security_config = response.get('SecurityConfiguration', {})

                success_message = f'Successfully retrieved security configuration {config_name}'
                data = GetSecurityConfigurationData(
                    config_name=security_config.get('Name', config_name),
                    config_details=security_config,
                    creation_time=(
                        response.get('CreatedTimeStamp', '').isoformat()
                        if response.get('CreatedTimeStamp')
                        else ''
                    ),
                    encryption_configuration=security_config.get('EncryptionConfiguration', {}),
                    operation='get-security-configuration',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            else:
                error_message = f'Invalid operation: {operation}. Must be one of: create-security-configuration, delete-security-configuration, get-security-configuration'
                log_with_request_id(ctx, LogLevel.ERROR, error_message)
                return CallToolResult(
                    isError=True,
                    content=[TextContent(type='text', text=error_message)],
                )

        except ValueError as e:
            log_with_request_id(ctx, LogLevel.ERROR, f'Parameter validation error: {str(e)}')
            raise
        except Exception as e:
            error_message = f'Error in manage_aws_glue_security: {str(e)}'
            log_with_request_id(ctx, LogLevel.ERROR, error_message)
            return CallToolResult(
                isError=True,
                content=[TextContent(type='text', text=error_message)],
            )

    async def manage_aws_glue_encryption(
        self,
        ctx: Context,
        operation: Annotated[
            str,
            Field(
                description='Operation to perform: get-catalog-encryption-settings, put-catalog-encryption-settings. Choose "get-catalog-encryption-settings" for read-only operations when write access is disabled.',
            ),
        ],
        catalog_id: Annotated[
            Optional[str],
            Field(
                description="ID of the Data Catalog to retrieve or update encryption settings for (defaults to caller's AWS account ID).",
            ),
        ] = None,
        encryption_at_rest: Annotated[
            Optional[Dict[str, Any]],
            Field(
                description='Encryption-at-rest configuration for the Data Catalog (for put-catalog-encryption-settings operation).',
            ),
        ] = None,
        connection_password_encryption: Annotated[
            Optional[Dict[str, Any]],
            Field(
                description='Connection password encryption configuration for the Data Catalog (for put-catalog-encryption-settings operation).',
            ),
        ] = None,
    ) -> CallToolResult:
        """Manage AWS Glue Data Catalog Encryption Settings for data protection.

        This tool allows you to retrieve and update AWS Glue Data Catalog Encryption Settings, which control
        how metadata and connection passwords are encrypted in the Data Catalog.

        ## Requirements
        - The server must be run with the `--allow-write` flag for put-catalog-encryption-settings operation
        - Appropriate AWS permissions for Glue Data Catalog Encryption operations

        ## Operations
        - **get-catalog-encryption-settings**: Retrieve the current encryption settings for the Data Catalog
        - **put-catalog-encryption-settings**: Update the encryption settings for the Data Catalog

        ## Example
        ```json
        {
          "operation": "put-catalog-encryption-settings",
          "encryption_at_rest": {
            "CatalogEncryptionMode": "SSE-KMS",
            "SseAwsKmsKeyId": "arn:aws:kms:region:account-id:key/key-id"
          },
          "connection_password_encryption": {
            "ReturnConnectionPasswordEncrypted": true,
            "AwsKmsKeyId": "arn:aws:kms:region:account-id:key/key-id"
          }
        }
        ```

        Args:
            ctx: MCP context
            operation: Operation to perform
            catalog_id: ID of the Data Catalog (optional, defaults to the caller's AWS account ID)
            encryption_at_rest: Encryption-at-rest configuration for the Data Catalog
            connection_password_encryption: Connection password encryption configuration

        Returns:
            Union of response types specific to the operation performed
        """
        try:
            if not self.allow_write and operation != 'get-catalog-encryption-settings':
                error_message = f'Operation {operation} is not allowed without write access'
                log_with_request_id(ctx, LogLevel.ERROR, error_message)

                return CallToolResult(
                    isError=True,
                    content=[TextContent(type='text', text=error_message)],
                )

            if operation == 'get-catalog-encryption-settings':
                # Prepare parameters
                params = {}
                if catalog_id:
                    params['CatalogId'] = catalog_id

                # Get the catalog encryption settings
                response = self.glue_client.get_data_catalog_encryption_settings(**params)

                success_message = 'Successfully retrieved Data Catalog encryption settings'
                data = GetDataCatalogEncryptionSettingsData(
                    encryption_settings=response.get('DataCatalogEncryptionSettings', {}),
                    operation='get-datacatalog-encryption',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            elif operation == 'put-catalog-encryption-settings':
                # Prepare encryption settings
                encryption_settings = {}
                if encryption_at_rest:
                    encryption_settings['EncryptionAtRest'] = encryption_at_rest
                if connection_password_encryption:
                    encryption_settings['ConnectionPasswordEncryption'] = (
                        connection_password_encryption
                    )

                if not encryption_settings:
                    raise ValueError(
                        'Either encryption_at_rest or connection_password_encryption is required for put-catalog-encryption-settings operation'
                    )

                # Prepare parameters
                params: Dict[str, Any] = {'DataCatalogEncryptionSettings': encryption_settings}
                if catalog_id:
                    params['CatalogId'] = catalog_id

                # Update the catalog encryption settings
                self.glue_client.put_data_catalog_encryption_settings(**params)

                success_message = 'Successfully updated Data Catalog encryption settings'
                data = PutDataCatalogEncryptionSettingsData(
                    operation='put-datacatalog-encryption',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            else:
                error_message = f'Invalid operation: {operation}. Must be one of: get-catalog-encryption-settings, put-catalog-encryption-settings'
                log_with_request_id(ctx, LogLevel.ERROR, error_message)
                return CallToolResult(
                    isError=True,
                    content=[TextContent(type='text', text=error_message)],
                )

        except ValueError as e:
            log_with_request_id(ctx, LogLevel.ERROR, f'Parameter validation error: {str(e)}')
            raise
        except Exception as e:
            error_message = f'Error in manage_aws_glue_encryption: {str(e)}'
            log_with_request_id(ctx, LogLevel.ERROR, error_message)
            return CallToolResult(
                isError=True,
                content=[TextContent(type='text', text=error_message)],
            )

    async def manage_aws_glue_resource_policies(
        self,
        ctx: Context,
        operation: Annotated[
            str,
            Field(
                description='Operation to perform: get-resource-policy, put-resource-policy, delete-resource-policy. Choose "get-resource-policy" for read-only operations when write access is disabled.',
            ),
        ],
        policy: Annotated[
            Optional[str],
            Field(
                description='Resource policy document for put-resource-policy operation.',
            ),
        ] = None,
        policy_hash: Annotated[
            Optional[str],
            Field(
                description='Hash of the policy to update or delete.',
            ),
        ] = None,
        policy_exists_condition: Annotated[
            Optional[str],
            Field(
                description='Condition under which to update or delete the policy (MUST_EXIST or NOT_EXIST).',
            ),
        ] = None,
        enable_hybrid: Annotated[
            Optional[bool],
            Field(
                description='Whether to enable hybrid access policy for put-resource-policy operation.',
            ),
        ] = None,
        resource_arn: Annotated[
            Optional[str],
            Field(
                description='ARN of the Glue resource for the resource policy (optional).',
            ),
        ] = None,
    ) -> CallToolResult:
        r"""Manage AWS Glue Resource Policies for access control.

        This tool allows you to retrieve, create, update, and delete AWS Glue Resource Policies, which
        control access to Glue resources through IAM policy documents.

        ## Requirements
        - The server must be run with the `--allow-write` flag for put-resource-policy and delete-resource-policy operations
        - Appropriate AWS permissions for Glue Resource Policy operations

        ## Operations
        - **get-resource-policy**: Retrieve the current resource policy
        - **put-resource-policy**: Create or update the resource policy
        - **delete-resource-policy**: Delete the resource policy

        ## Example
        ```json
        {
          "operation": "put-resource-policy",
          "policy": "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"AWS\":\"arn:aws:iam::123456789…
          "policy_exists_condition": "NOT_EXIST",
          "enable_hybrid": true
        }
        ```

        Args:
            ctx: MCP context
            operation: Operation to perform
            policy: Resource policy document for put-resource-policy operation
            policy_hash: Hash of the policy to update or delete
            policy_exists_condition: Condition under which to update or delete the policy
            enable_hybrid: Whether to enable hybrid access policy for put-resource-policy operation
            resource_arn: ARN of the Glue resource for the resource policy

        Returns:
            Union of response types specific to the operation performed
        """
        try:
            if not self.allow_write and operation != 'get-resource-policy':
                error_message = f'Operation {operation} is not allowed without write access'
                log_with_request_id(ctx, LogLevel.ERROR, error_message)

                return CallToolResult(
                    isError=True,
                    content=[TextContent(type='text', text=error_message)],
                )

            if operation == 'get-resource-policy':
                # Prepare parameters
                params = {}
                if resource_arn:
                    params['ResourceArn'] = resource_arn

                # Get the resource policy
                response = self.glue_client.get_resource_policy(**params)

                success_message = 'Successfully retrieved resource policy'
                data = GetResourcePolicyData(
                    policy_hash=response.get('PolicyHash'),
                    policy_in_json=response.get('PolicyInJson'),
                    create_time=(
                        response.get('CreateTime', '').isoformat()
                        if response.get('CreateTime')
                        else None
                    ),
                    update_time=(
                        response.get('UpdateTime', '').isoformat()
                        if response.get('UpdateTime')
                        else None
                    ),
                    operation='get-resource-policy',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            elif operation == 'put-resource-policy':
                if policy is None:
                    raise ValueError('policy is required for put-resource-policy operation')

                # Prepare parameters
                params: Dict[str, Any] = {'PolicyInJson': policy}
                if policy_hash:
                    params['PolicyHashCondition'] = policy_hash
                if policy_exists_condition:
                    params['PolicyExistsCondition'] = policy_exists_condition
                if enable_hybrid is not None:
                    params['EnableHybrid'] = enable_hybrid
                if resource_arn:
                    params['ResourceArn'] = resource_arn

                # Update the resource policy
                response = self.glue_client.put_resource_policy(**params)

                success_message = 'Successfully updated resource policy'
                data = PutResourcePolicyData(
                    policy_hash=response.get('PolicyHash'),
                    operation='put-resource-policy',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            elif operation == 'delete-resource-policy':
                # Prepare parameters
                params = {}
                if policy_hash:
                    params['PolicyHashCondition'] = policy_hash
                if resource_arn:
                    params['ResourceArn'] = resource_arn

                # Delete the resource policy
                self.glue_client.delete_resource_policy(**params)

                success_message = 'Successfully deleted resource policy'
                data = DeleteResourcePolicyData(
                    operation='delete-resource-policy',
                )

                return CallToolResult(
                    isError=False,
                    content=[
                        TextContent(type='text', text=success_message),
                        TextContent(type='text', text=data.model_dump_json()),
                    ],
                )

            else:
                error_message = f'Invalid operation: {operation}. Must be one of: get-resource-policy, put-resource-policy, delete-resource-policy'
                log_with_request_id(ctx, LogLevel.ERROR, error_message)
                return CallToolResult(
                    isError=True,
                    content=[TextContent(type='text', text=error_message)],
                )

        except ValueError as e:
            log_with_request_id(ctx, LogLevel.ERROR, f'Parameter validation error: {str(e)}')
            raise
        except Exception as e:
            error_message = f'Error in manage_aws_glue_resource_policies: {str(e)}'
            log_with_request_id(ctx, LogLevel.ERROR, error_message)
            return CallToolResult(
                isError=True,
                content=[TextContent(type='text', text=error_message)],
            )
