turingears/agent_engine_dlp repository overview

FROM google/cloud-sdk:slim
LABEL maintainer="TurinGears"
LABEL version="1.0.0"
LABEL description="Docker image for Agent Engine"
ENV PIP_BREAK_SYSTEM_PACKAGES=1
WORKDIR /app
COPY requirements.txt ./
COPY deploy_to_agent_engine.py ./
RUN pip3 install -r requirements.txt --no-cache-dir

google-cloud-aiplatform[adk,agent_engines]
python-dotenv
toolbox-core
vertexai

"""
Deploy Agent to Vertex AI Agent Engine
"""
"""
Deploy Agent to Vertex AI Agent Engine
"""
import os
import sys
import argparse
import importlib
import vertexai
from vertexai import agent_engines
from vertexai.preview import reasoning_engines

def parse_arguments():
    parser = argparse.ArgumentParser(description='Deploy Agent to Vertex AI Agent Engine')
    
    parser.add_argument('--agent-dir', required=True, 
                       help='Agent directory name (e.g., cultura_agent, energy_agent)')
    parser.add_argument('--project-id', required=True,
                       help='Google Cloud Project ID')
    parser.add_argument('--location', default='us-central1',
                       help='Google Cloud Location (default: us-central1)')
    parser.add_argument('--staging-bucket', required=True,
                       help='Staging bucket for deployment (e.g., gs://my-bucket)')
    parser.add_argument('--display-name', required=True,
                       help='Display name for the agent')
    parser.add_argument('--description', required=True,
                       help='Description of the agent')

    
    return parser.parse_args()

def load_requirements(agent_dir):
    """Load requirements from requirements.txt file"""
    # El requirements.txt está en el directorio padre del agent_dir
    requirements_path = os.path.join(os.path.dirname(agent_dir), 'requirements.txt')
    
    if not os.path.exists(requirements_path):
        print(f"⚠️  Warning: {requirements_path} not found, using default requirements")
        # Fallback a requirements por defecto
        return [
            "google-cloud-aiplatform[adk,agent_engines]",
            "python-dotenv",
            "pydantic-settings",
            "toolbox-core"
        ]
    
    try:
        with open(requirements_path, 'r', encoding='utf-8') as f:
            requirements = []
            for line in f:
                line = line.strip()
                # Ignorar líneas vacías y comentarios
                if line and not line.startswith('#'):
                    requirements.append(line)
        
        print(f"✅ Loaded {len(requirements)} requirements from {requirements_path}")
        return requirements
        
    except Exception as e:
        print(f"❌ Error reading {requirements_path}: {e}")
        print("Using default requirements...")
        return [
            "google-cloud-aiplatform[adk,agent_engines]",
            "python-dotenv",
            "pydantic-settings",
            "toolbox-core"
        ]

def load_env_vars(agent_dir):
    """Load environment variables from .env file, ignoring project/location variables"""
    env_path = os.path.join(agent_dir, '.env')
    
    if not os.path.exists(env_path):
        print(f"⚠️  Warning: {env_path} not found, no environment variables loaded")
        return None
    
    env_vars = {}
    ignored_vars = {"GOOGLE_CLOUD_PROJECT", "GOOGLE_CLOUD_LOCATION"}
    
    try:
        with open(env_path, 'r', encoding='utf-8') as f:
            for line_num, line in enumerate(f, 1):
                line = line.strip()
                
                # Ignorar líneas vacías y comentarios
                if not line or line.startswith('#'):
                    continue
                
                # Buscar formato KEY=VALUE
                if '=' in line:
                    key, value = line.split('=', 1)
                    key = key.strip()
                    value = value.strip()
                    
                    if key in ignored_vars:
                        print(f"⚠️  Ignoring environment variable {key}")
                        continue
                    
                    # Remover comillas si existen
                    if value.startswith('"') and value.endswith('"'):
                        value = value[1:-1]
                    elif value.startswith("'") and value.endswith("'"):
                        value = value[1:-1]
                    
                    env_vars[key] = value
                else:
                    print(f"⚠️  Skipping invalid line {line_num} in {env_path}: {line}")
        
        print(f"✅ Loaded {len(env_vars)} environment variables from {env_path}")
        
        # Mostrar variables cargadas (sin valores por seguridad)
        for key in env_vars.keys():
            print(f"   📌 {key}")
        
        return env_vars if env_vars else None
        
    except Exception as e:
        print(f"❌ Error reading {env_path}: {e}")
        return None

def load_agent(agent_dir):
    """Dynamically load the root_agent from the specified directory"""
    try:
        module_path = f"{agent_dir}.agent"
        agent_module = importlib.import_module(module_path)
        return agent_module.root_agent
    except ImportError as e:
        print(f"❌ Error importing agent from {module_path}: {e}")
        sys.exit(1)

def main():
    args = parse_arguments()
    
    # Verificar que el directorio del agente existe
    if not os.path.isdir(args.agent_dir):
        print(f"❌ Agent directory not found: {args.agent_dir}")
        sys.exit(1)
    
    print(f"🚀 Preparing deployment for {args.agent_dir}...")
    
    # Load requirements from file
    requirements = load_requirements(args.agent_dir)
    print(f"📦 Requirements: {len(requirements)} packages")
    
    # Load environment variables from .env
    env_vars = load_env_vars(args.agent_dir)
    
    # Load the agent dynamically
    print(f"🤖 Loading agent from {args.agent_dir}...")
    root_agent = load_agent(args.agent_dir)
    
    # Initialize Vertex AI
    print(f"☁️  Initializing Vertex AI with project: {args.project_id}, location: {args.location}")
    vertexai.init(
        project=args.project_id,
        location=args.location,
        staging_bucket=f'''gs://{args.staging_bucket}''',
    )
    

    
    print(f"\n🚀 Deploying {args.agent_dir} to Vertex AI Agent Engine...")
    print(f"📋 Display Name: {args.display_name}")
    print(f"📝 Description: {args.description}")
    
    # Wrap the agent for Agent Engine
    app = reasoning_engines.AdkApp(
        agent=root_agent,
        enable_tracing=True,
    )
    
    # Prepare deployment arguments
    deploy_args = {
        'display_name': args.display_name,
        'description': args.description,
        'requirements': requirements,
        'extra_packages': [f"./{args.agent_dir}/"],
    }
    
    # Add environment variables if loaded
    if env_vars:
        deploy_args['env_vars'] = env_vars
        print(f"🔧 Including {len(env_vars)} environment variables")
    
    # Deploy to Agent Engine
    print("⏳ Starting deployment (this may take several minutes)...")
    remote_agent = agent_engines.create(app, **deploy_args)
    
    print(f"\n✅ Agent deployed successfully!")
    print(f"📋 Resource name: {remote_agent.resource_name}")
    print(f"🏷️  Display name: {args.display_name}")
    print(f"📁 Agent directory: {args.agent_dir}")
    print(f"📦 Requirements loaded from: requirements.txt (root directory)")
    if env_vars:
        print(f"🔧 Environment variables loaded from: {args.agent_dir}/.env")

    os._exit(0)

if __name__ == "__main__":
    main()