Setting the file. One moment.
Launch API Client · Launch With AWS · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 69
Creating Amazon Aurora Db Cluster With Instances
92
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Debugging Lambda Timeouts
def _client_error_to_api_error
— line 162
This file
Number 45.5
Position 5 of 7
Type Python
Size 10 KB
Lines 297 scripts/ launch_api_client.py
Python · 297 lines · 10 KB
urllib.request
16 from pathlib import Path
17 from typing import Any, Dict, Optional
18
19 import boto3
20 import botocore.loaders
21 import botocore.session
22 import botocore.tokens
23 from auth import get_access_token
24 from botocore.config import Config as BotoConfig
25 from botocore.exceptions import ClientError
26 from launch_config import (
27 DEFAULT_BASE_URL ,
28 REQUEST_TIMEOUT_SECS ,
29 UPLOAD_TIMEOUT_SECS ,
30 load_config,
31 )
32
33 # ── Boto3 custom service client ──────────────────────────────────────────
34
35 _SERVICE_MODEL_PATH = Path( __file__ ).parent.parent / "references" / "launchwithaws-2026-06-15.json"
36
37 SERVICE_NAME = "launchwithaws"
38 API_VERSION = "2026-06-15"
39
40 logger = logging.getLogger( __name__ )
41
42 # Replace UUIDs and long id-like runs in a request path with a placeholder.
43 _UUID_RE = re.compile(
44 r " [ 0-9a-fA-F ] {8} - [ 0-9a-fA-F ] {4} - [ 0-9a-fA-F ] {4} - [ 0-9a-fA-F ] {4} - [ 0-9a-fA-F ] {12} "
45 )
46 _ID_SEGMENT_RE = re.compile( r " [ A-Za-z0-9_- ] {16,} " )
47
48
49 def _sanitize_path (path: str ) -> str :
50 """Strip query strings and replace id-like path segments with <id>."""
51 path = path.split( "?" , 1 )[ 0 ].split( "#" , 1 )[ 0 ]
52 path = _UUID_RE .sub( "<id>" , path)
53 path = _ID_SEGMENT_RE .sub( "<id>" , path)
54 return path
55
56
57 def _load_service_model () -> dict :
58 with open ( _SERVICE_MODEL_PATH , "r" ) as f:
59 return json.load(f)
60
61
62 def _register_service (session: botocore.session.Session) -> None :
63 model = _load_service_model()
64 original_load_service_model = session.get_component( "data_loader" ).load_service_model
65
66 def patched_load_service_model (service_name, type_name, api_version = None ):
67 if service_name == SERVICE_NAME and type_name == "service-2" :
68 return model
69 return original_load_service_model(service_name, type_name, api_version)
70
71 session.get_component( "data_loader" ).load_service_model = patched_load_service_model
72
73 original_list = session.get_component( "data_loader" ).list_available_services
74
75 def patched_list (type_name = "service-2" ):
76 services = original_list(type_name)
77 if type_name == "service-2" and SERVICE_NAME not in services:
78 services.append( SERVICE_NAME )
79 return services
80
81 session.get_component( "data_loader" ).list_available_services = patched_list
82
83
84 class _StaticTokenProvider :
85 METHOD = "static"
86
87 def __init__ (self, token: str ) -> None :
88 self ._token = botocore.tokens.FrozenAuthToken( token = token, expiration = None )
89
90 def load_token (self, ** kwargs) -> botocore.tokens.FrozenAuthToken:
91 return self ._token
92
93
94 def create_client (
95 endpoint_url: Optional[ str ] = None ,
96 bearer_token: Optional[ str ] = None ,
97 region_name: str = "us-east-1" ,
98 config: Optional[BotoConfig] = None ,
99 ):
100 """Create a boto3 client for the Launch with AWS service."""
101 if endpoint_url is None :
102 endpoint_url = os.environ.get( "LAUNCH_WITH_AWS_BASE_URL" , DEFAULT_BASE_URL )
103
104 botocore_session = botocore.session.get_session()
105 _register_service(botocore_session)
106
107 if bearer_token:
108 static_provider = _StaticTokenProvider(bearer_token)
109 token_chain = botocore.tokens.TokenProviderChain( providers = [static_provider])
110 botocore_session.register_component( "token_provider" , token_chain)
111
112 boto3_session = boto3.Session( botocore_session = botocore_session, region_name = region_name)
113
114 client_config = config or BotoConfig(
115 retries = { "max_attempts" : 3 , "mode" : "adaptive" },
116 )
117
118 return boto3_session.client(
119 SERVICE_NAME ,
120 endpoint_url = endpoint_url,
121 config = client_config,
122 aws_access_key_id = "unused" ,
123 aws_secret_access_key = "unused" ,
124 )
125
126
127 # ── API client ───────────────────────────────────────────────────────────
128
129
130 class ApiError ( Exception ):
131 """Raised when a backend API request returns a non-success status.
132
133 The string representation exposes only the HTTP method, status, a sanitized
134 path template, and the structured error code. The raw response body is kept
135 on the instance (``body``) and logged at DEBUG for local diagnosis.
136 """
137
138 def __init__ (
139 self,
140 status: int ,
141 method: str ,
142 path: str ,
143 body: str ,
144 error_code: Optional[ str ] = None ,
145 ) -> None :
146 self .status = status
147 self .method = method
148 self .path = path
149 self .body = body
150 self .error_code = error_code
151
152 safe_path = _sanitize_path(path)
153 if error_code:
154 message = f " { method } { safe_path } failed ( { status } ): { error_code } "
155 else :
156 message = f " { method } { safe_path } failed ( { status } )"
157 super (). __init__ (message)
158
159 logger.debug( "API error %s %s ( %s ): %s " , method, path, status, body)
160
161
162 def _client_error_to_api_error (err: ClientError, method: str , path: str ) -> ApiError:
163 status = err.response.get( "ResponseMetadata" , {}).get( "HTTPStatusCode" , 500 )
164 error_code = err.response.get( "Error" , {}).get( "Code" )
165 raw_message = err.response.get( "Error" , {}).get( "Message" , str (err))
166 return ApiError(status, method, path, raw_message, error_code = error_code)
167
168
169 def _get_base_url () -> str :
170 return load_config().base_url
171
172
173 def _get_boto3_client () -> Any:
174 """Create a boto3 client with the current Bearer token."""
175 token = get_access_token()
176 return create_client(
177 endpoint_url = _get_base_url(),
178 bearer_token = token,
179 config = BotoConfig(
180 retries = { "max_attempts" : 3 , "mode" : "adaptive" },
181 connect_timeout = REQUEST_TIMEOUT_SECS ,
182 read_timeout = REQUEST_TIMEOUT_SECS ,
183 ),
184 )
185
186
187 # ── Public API functions ─────────────────────────────────────────────────
188
189
190 def create_upload_url () -> Any:
191 """POST /api/uploads -> presigned S3 PUT location."""
192 client = _get_boto3_client()
193 try :
194 return client.create_upload_url()
195 except ClientError as err:
196 raise _client_error_to_api_error(err, "POST" , "/api/uploads" ) from err
197
198
199 def put_archive (upload_url: str , archive: bytes ) -> None :
200 """Upload zip bytes to a presigned S3 PUT URL."""
201 req = urllib.request.Request(
202 upload_url,
203 data = archive,
204 headers = { "Content-Type" : "application/zip" },
205 method = "PUT" ,
206 )
207 try :
208 with urllib.request.urlopen(req, timeout = UPLOAD_TIMEOUT_SECS ):
209 pass
210 except urllib.error.HTTPError as err:
211 raise ApiError(err.code, "PUT" , "(presigned S3 upload)" , err.read().decode()) from err
212
213
214 # ── Launch resource API ──────────────────────────────────────────────────
215
216
217 def create_launch (
218 name: str ,
219 source: Dict[ str , Any],
220 client_token: Optional[ str ] = None ,
221 ) -> Any:
222 """POST /api/launches — create a new launch from an upload or GitHub repo."""
223 client = _get_boto3_client()
224 kwargs: Dict[ str , Any] = { "name" : name, "source" : source}
225 if client_token:
226 kwargs[ "clientToken" ] = client_token
227 try :
228 return client.create_launch( ** kwargs)
229 except ClientError as err:
230 raise _client_error_to_api_error(err, "POST" , "/api/launches" ) from err
231
232
233 def get_launch (launch_id: str , include: Optional[ str ] = None ) -> Any:
234 """GET /api/launches/:launchId — get launch details with optional sections."""
235 client = _get_boto3_client()
236 kwargs: Dict[ str , Any] = { "launchIdentifier" : launch_id}
237 if include:
238 # The API accepts a list of section enum values.
239 kwargs[ "include" ] = [s.strip() for s in include.split( "," )]
240 try :
241 return client.get_launch( ** kwargs)
242 except ClientError as err:
243 raise _client_error_to_api_error(err, "GET" , f "/api/launches/ { launch_id } " ) from err
244
245
246 def list_launches (max_results: Optional[ int ] = None ) -> Any:
247 """GET /api/launches — list all launches for the current user."""
248 client = _get_boto3_client()
249 kwargs: Dict[ str , Any] = {}
250 if max_results is not None :
251 kwargs[ "maxResults" ] = max_results
252 try :
253 return client.list_launches( ** kwargs)
254 except ClientError as err:
255 raise _client_error_to_api_error(err, "GET" , "/api/launches" ) from err
256
257
258 def delete_launch (launch_id: str ) -> None :
259 """DELETE /api/launches/:launchId — delete a launch (returns 204 No Content)."""
260 client = _get_boto3_client()
261 try :
262 client.delete_launch( launchIdentifier = launch_id)
263 except ClientError as err:
264 raise _client_error_to_api_error(err, "DELETE" , f "/api/launches/ { launch_id } " ) from err
265
266
267 def refine_plan (
268 launch_id: str ,
269 context_answers: Optional[Dict[ str , str ]] = None ,
270 prompt: Optional[ str ] = None ,
271 client_token: Optional[ str ] = None ,
272 ) -> Any:
273 """POST /api/launches/:launchId/refine — provide context answers to refine the plan."""
274 client = _get_boto3_client()
275 kwargs: Dict[ str , Any] = { "launchIdentifier" : launch_id}
276 if context_answers:
277 kwargs[ "contextAnswers" ] = context_answers
278 if prompt:
279 kwargs[ "prompt" ] = prompt
280 if client_token:
281 kwargs[ "clientToken" ] = client_token
282 try :
283 return client.refine_plan( ** kwargs)
284 except ClientError as err:
285 raise _client_error_to_api_error(err, "POST" , f "/api/launches/ { launch_id } /refine" ) from err
286
287
288 def start_launch_execution (launch_id: str , client_token: Optional[ str ] = None ) -> Any:
289 """POST /api/launches/:launchId/start — start execution of the deployment plan."""
290 client = _get_boto3_client()
291 kwargs: Dict[ str , Any] = { "launchIdentifier" : launch_id}
292 if client_token:
293 kwargs[ "clientToken" ] = client_token
294 try :
295 return client.start_launch_execution( ** kwargs)
296 except ClientError as err:
297 raise _client_error_to_api_error(err, "POST" , f "/api/launches/ { launch_id } /start" ) from err