repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources16d ago| #1 | import os |
| #2 | import json |
| #3 | from typing import Optional, Dict, Any |
| #4 | |
| #5 | try: |
| #6 | from google.oauth2 import service_account |
| #7 | from google.auth import default |
| #8 | import google.auth.credentials |
| #9 | except ImportError: |
| #10 | raise ImportError("google-auth is required for GCP authentication. Install with: pip install google-auth") |
| #11 | |
| #12 | |
| #13 | class GCPAuthenticator: |
| #14 | """ |
| #15 | Centralized GCP authentication handler that supports multiple credential methods. |
| #16 | |
| #17 | Priority order: |
| #18 | 1. service_account_json (dict) - In-memory service account credentials |
| #19 | 2. credentials_path (str) - Path to service account JSON file |
| #20 | 3. Environment variables (GOOGLE_APPLICATION_CREDENTIALS) |
| #21 | 4. Default credentials (for environments like GCE, Cloud Run, etc.) |
| #22 | """ |
| #23 | |
| #24 | @staticmethod |
| #25 | def get_credentials( |
| #26 | service_account_json: Optional[Dict[str, Any]] = None, |
| #27 | credentials_path: Optional[str] = None, |
| #28 | scopes: Optional[list] = None |
| #29 | ) -> tuple[google.auth.credentials.Credentials, Optional[str]]: |
| #30 | """ |
| #31 | Get Google credentials using the priority order defined above. |
| #32 | |
| #33 | Args: |
| #34 | service_account_json: Service account credentials as a dictionary |
| #35 | credentials_path: Path to service account JSON file |
| #36 | scopes: List of OAuth scopes (optional) |
| #37 | |
| #38 | Returns: |
| #39 | tuple: (credentials, project_id) |
| #40 | |
| #41 | Raises: |
| #42 | ValueError: If no valid credentials are found |
| #43 | """ |
| #44 | credentials = None |
| #45 | project_id = None |
| #46 | |
| #47 | # Method 1: Service account JSON (in-memory) |
| #48 | if service_account_json: |
| #49 | credentials = service_account.Credentials.from_service_account_info( |
| #50 | service_account_json, scopes=scopes |
| #51 | ) |
| #52 | project_id = service_account_json.get("project_id") |
| #53 | |
| #54 | # Method 2: Service account file path |
| #55 | elif credentials_path and os.path.isfile(credentials_path): |
| #56 | credentials = service_account.Credentials.from_service_account_file( |
| #57 | credentials_path, scopes=scopes |
| #58 | ) |
| #59 | # Extract project_id from the file |
| #60 | with open(credentials_path, 'r') as f: |
| #61 | cred_data = json.load(f) |
| #62 | project_id = cred_data.get("project_id") |
| #63 | |
| #64 | # Method 3: Environment variable path |
| #65 | elif os.getenv("GOOGLE_APPLICATION_CREDENTIALS"): |
| #66 | env_path = os.getenv("GOOGLE_APPLICATION_CREDENTIALS") |
| #67 | if os.path.isfile(env_path): |
| #68 | credentials = service_account.Credentials.from_service_account_file( |
| #69 | env_path, scopes=scopes |
| #70 | ) |
| #71 | # Extract project_id from the file |
| #72 | with open(env_path, 'r') as f: |
| #73 | cred_data = json.load(f) |
| #74 | project_id = cred_data.get("project_id") |
| #75 | |
| #76 | # Method 4: Default credentials (GCE, Cloud Run, etc.) |
| #77 | if not credentials: |
| #78 | try: |
| #79 | credentials, project_id = default(scopes=scopes) |
| #80 | except Exception as e: |
| #81 | raise ValueError( |
| #82 | f"No valid GCP credentials found. Please provide one of:\n" |
| #83 | f"1. service_account_json parameter (dict)\n" |
| #84 | f"2. credentials_path parameter (file path)\n" |
| #85 | f"3. GOOGLE_APPLICATION_CREDENTIALS environment variable\n" |
| #86 | f"4. Default credentials (if running on GCP)\n" |
| #87 | f"Error: {e}" |
| #88 | ) |
| #89 | |
| #90 | return credentials, project_id |
| #91 | |
| #92 | @staticmethod |
| #93 | def setup_vertex_ai( |
| #94 | service_account_json: Optional[Dict[str, Any]] = None, |
| #95 | credentials_path: Optional[str] = None, |
| #96 | project_id: Optional[str] = None, |
| #97 | location: str = "us-central1" |
| #98 | ) -> str: |
| #99 | """ |
| #100 | Initialize Vertex AI with proper authentication. |
| #101 | |
| #102 | Args: |
| #103 | service_account_json: Service account credentials as dict |
| #104 | credentials_path: Path to service account JSON file |
| #105 | project_id: GCP project ID (optional, will be auto-detected) |
| #106 | location: GCP location/region |
| #107 | |
| #108 | Returns: |
| #109 | str: The project ID being used |
| #110 | |
| #111 | Raises: |
| #112 | ValueError: If authentication fails |
| #113 | """ |
| #114 | try: |
| #115 | import vertexai |
| #116 | except ImportError: |
| #117 | raise ImportError("google-cloud-aiplatform is required for Vertex AI. Install with: pip install google-cloud-aiplatform") |
| #118 | |
| #119 | credentials, detected_project_id = GCPAuthenticator.get_credentials( |
| #120 | service_account_json=service_account_json, |
| #121 | credentials_path=credentials_path, |
| #122 | scopes=["https://www.googleapis.com/auth/cloud-platform"] |
| #123 | ) |
| #124 | |
| #125 | # Use provided project_id or fall back to detected one |
| #126 | final_project_id = project_id or detected_project_id or os.getenv("GOOGLE_CLOUD_PROJECT") |
| #127 | |
| #128 | if not final_project_id: |
| #129 | raise ValueError("Project ID could not be determined. Please provide project_id parameter or set GOOGLE_CLOUD_PROJECT environment variable.") |
| #130 | |
| #131 | vertexai.init(project=final_project_id, location=location, credentials=credentials) |
| #132 | return final_project_id |
| #133 | |
| #134 | @staticmethod |
| #135 | def get_genai_client( |
| #136 | service_account_json: Optional[Dict[str, Any]] = None, |
| #137 | credentials_path: Optional[str] = None, |
| #138 | api_key: Optional[str] = None |
| #139 | ): |
| #140 | """ |
| #141 | Get a Google GenAI client with authentication. |
| #142 | |
| #143 | Args: |
| #144 | service_account_json: Service account credentials as dict |
| #145 | credentials_path: Path to service account JSON file |
| #146 | api_key: API key (takes precedence over service account) |
| #147 | |
| #148 | Returns: |
| #149 | Google GenAI client instance |
| #150 | """ |
| #151 | try: |
| #152 | from google.genai import Client as GenAIClient |
| #153 | except ImportError: |
| #154 | raise ImportError("google-genai is required. Install with: pip install google-genai") |
| #155 | |
| #156 | # If API key is provided, use it directly |
| #157 | if api_key: |
| #158 | return GenAIClient(api_key=api_key) |
| #159 | |
| #160 | # Otherwise, try service account authentication |
| #161 | credentials, _ = GCPAuthenticator.get_credentials( |
| #162 | service_account_json=service_account_json, |
| #163 | credentials_path=credentials_path, |
| #164 | scopes=["https://www.googleapis.com/auth/generative-language"] |
| #165 | ) |
| #166 | |
| #167 | return GenAIClient(credentials=credentials) |