Setting the file. One moment.
Launch With AWS · 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
Creating API Gateway Stage
def cmd_refine_plan
— line 145
This file
Number 45.7
Position 7 of 7
Type Python
Size 8 KB
Lines 267 scripts/ launch_with_aws.py
Python · 267 lines · 8 KB
17 import sys
18 from pathlib import Path
19 from urllib.parse import urlparse
20
21 logger = logging.getLogger( __name__ )
22
23 # The scripts use PEP 604 union syntax (`str | None`) in signatures, which
24 # fails at definition time on Python < 3.10; boto3 also requires 3.10+.
25 if sys.version_info < ( 3 , 10 ):
26 print (
27 f "Error: Python 3.10+ required "
28 f "(found { sys.version_info.major } . { sys.version_info.minor } )" ,
29 file = sys.stderr,
30 )
31 sys.exit( 2 )
32
33 try :
34 import boto3 # noqa: F401
35 except ImportError :
36 print (
37 "Error: missing required package: boto3 \n " "Install it with: \n " " pip install boto3" ,
38 file = sys.stderr,
39 )
40 sys.exit( 2 )
41
42 # Ensure sibling modules are importable regardless of cwd.
43 sys.path.insert( 0 , str (Path( __file__ ).parent))
44
45 import launch_api_client as api
46 from archive import ArchiveError, parse_github_url, zip_local_repo
47 from auth import SessionExpiredError, session_status, sign_out, start_auth, wait_for_auth
48 from launch_config import load_config
49
50
51 def _ok (value) -> None :
52 """Print a JSON result and exit 0."""
53 print (json.dumps(value, indent = 2 , default = str ))
54 sys.exit( 0 )
55
56
57 def _fail (message: str ) -> None :
58 """Print error to stderr and exit 1."""
59 print (json.dumps({ "error" : message}), file = sys.stderr)
60 sys.exit( 1 )
61
62
63 def _default_repo_name (source: str ) -> str :
64 github = parse_github_url(source)
65 if github:
66 return github[ 1 ]
67 cleaned = source.rstrip( "/" ).rstrip(os.sep)
68 return os.path.basename(cleaned) or "uploaded-app"
69
70
71 # ── Subcommands ──────────────────────────────────────────────────────────
72
73
74 def cmd_auth_start () -> None :
75 """Start authentication (non-blocking)."""
76 config = load_config()
77 result = start_auth()
78 result[ "baseUrl" ] = config.base_url
79 _ok(result)
80
81
82 def cmd_auth_wait (pid: str ) -> None :
83 """Wait for interactive authentication to complete."""
84 config = load_config()
85 result = wait_for_auth( pid = int (pid))
86 result[ "baseUrl" ] = config.base_url
87 _ok(result)
88
89
90 def cmd_session_status () -> None :
91 """Report the local session state without triggering authentication."""
92 _ok(session_status())
93
94
95 def cmd_sign_out () -> None :
96 """Sign out and delete the local session; requires re-auth next use."""
97 _ok(sign_out())
98
99
100 def cmd_create_launch (source: str , name: str | None = None ) -> None :
101 """Create a launch from a local path or GitHub URL.
102
103 For local paths, zips and uploads first. For GitHub URLs, passes directly.
104 """
105 display_name = (name or "" ).strip() or _default_repo_name(source)
106
107 github = parse_github_url(source)
108 if github:
109 # GitHub URL — pass as gitHub source directly.
110 launch_source = { "gitHub" : { "repositoryUrl" : source}}
111 elif urlparse(source).scheme in ( "http" , "https" ):
112 _fail(
113 f "Unsupported repository URL: { source } . Provide a "
114 "https://github.com/owner/name URL or a local directory path."
115 )
116 return
117 else :
118 # Local directory — zip, upload, then pass as s3Upload source.
119 archive = zip_local_repo(source, display_name)
120 target = api.create_upload_url()
121 api.put_archive(target[ "uploadUrl" ], archive)
122 launch_source = { "s3Upload" : { "uploadId" : target[ "uploadId" ]}}
123
124 result = api.create_launch( name = display_name, source = launch_source)
125 _ok(result.get( "launch" , result))
126
127
128 def cmd_get_launch (launch_id: str , include: str | None = None ) -> None :
129 """Get launch details, optionally including specific sections."""
130 result = api.get_launch(launch_id, include = include)
131 _ok(result.get( "launch" , result))
132
133
134 def cmd_list_launches () -> None :
135 """List all launches for the current user."""
136 _ok(api.list_launches())
137
138
139 def cmd_delete_launch (launch_id: str ) -> None :
140 """Delete a launch."""
141 api.delete_launch(launch_id)
142 _ok({ "deleted" : True , "id" : launch_id})
143
144
145 def cmd_refine_plan (launch_id: str , * context_pairs: str ) -> None :
146 """Refine a launch plan with context answers (key=value pairs)."""
147 context_answers = {}
148 for pair in context_pairs:
149 if "=" in pair:
150 key, value = pair.split( "=" , 1 )
151 context_answers[key.strip()] = value.strip()
152 _ok(api.refine_plan(launch_id, context_answers = context_answers or None ).get( "launch" , {}))
153
154
155 def cmd_start_launch_execution (launch_id: str ) -> None :
156 """Start execution of a launch's deployment plan."""
157 _ok(api.start_launch_execution(launch_id).get( "launch" , {}))
158
159
160 def cmd_get_launch_status (launch_id: str ) -> None :
161 """Poll launch status including execution progress."""
162 raw = api.get_launch(launch_id, include = "execution,cost_estimate" )
163 result = raw.get( "launch" , raw)
164 status = result.get( "status" )
165 execution = result.get( "execution" )
166
167 output = {
168 "id" : result.get( "id" ),
169 "status" : status,
170 "isComplete" : status == "completed" ,
171 "isFailed" : status == "failed" ,
172 }
173
174 if execution:
175 output[ "completedTasks" ] = execution.get( "completedTasks" )
176 output[ "totalTasks" ] = execution.get( "totalTasks" )
177 output[ "currentPhase" ] = execution.get( "currentPhase" )
178
179 if result.get( "costEstimate" ):
180 output[ "costEstimate" ] = result[ "costEstimate" ]
181
182 if result.get( "failureReason" ):
183 output[ "failureReason" ] = result[ "failureReason" ]
184
185 if result.get( "contextInputs" ):
186 output[ "contextInputs" ] = result[ "contextInputs" ]
187
188 _ok(output)
189
190
191 def cmd_get_launch_download_url (launch_id: str ) -> None :
192 """Get the download URL for a completed launch."""
193 raw = api.get_launch(launch_id, include = "download_url" )
194 result = raw.get( "launch" , raw)
195 download_url = result.get( "downloadUrl" )
196 if not download_url:
197 _fail( "Download URL not available yet. Ensure the launch execution has completed." )
198 return
199 _ok({ "downloadUrl" : download_url})
200
201
202 # ── CLI dispatcher ───────────────────────────────────────────────────────
203
204 # (func, min_required_args)
205 from typing import Any, Callable
206
207 COMMANDS : dict[ str , tuple[Callable[ ... , Any], int ]] = {
208 "auth-start" : (cmd_auth_start, 0 ),
209 "auth-wait" : (cmd_auth_wait, 1 ),
210 "session-status" : (cmd_session_status, 0 ),
211 "sign-out" : (cmd_sign_out, 0 ),
212 "create-launch" : (cmd_create_launch, 1 ),
213 "get-launch" : (cmd_get_launch, 1 ),
214 "list-launches" : (cmd_list_launches, 0 ),
215 "delete-launch" : (cmd_delete_launch, 1 ),
216 "refine-plan" : (cmd_refine_plan, 1 ),
217 "start-launch-execution" : (cmd_start_launch_execution, 1 ),
218 "get-launch-status" : (cmd_get_launch_status, 1 ),
219 "get-launch-download-url" : (cmd_get_launch_download_url, 1 ),
220 }
221
222
223 def _usage () -> str :
224 return "Usage: launch_with_aws.py <command> [args...] \n\n " "Commands: \n " + " \n " .join(
225 f " { name } " for name in COMMANDS
226 )
227
228
229 def main () -> None :
230 args = sys.argv[ 1 :]
231 if not args or args[ 0 ] in ( "-h" , "--help" ):
232 print (_usage())
233 sys.exit( 0 )
234
235 command = args[ 0 ]
236 if command not in COMMANDS :
237 print ( f "Unknown command: { command }\n\n{ _usage() } " , file = sys.stderr)
238 sys.exit( 1 )
239
240 func, min_args = COMMANDS [command]
241 cmd_args = args[ 1 :]
242
243 if len (cmd_args) < min_args:
244 _fail( f "Missing required argument for { command } " )
245
246 try :
247 func( * cmd_args)
248 except SessionExpiredError as err:
249 _fail( str (err))
250 except ArchiveError as err:
251 _fail( str (err))
252 except api.ApiError as err:
253 hint = ""
254 if err.status == 401 :
255 hint = (
256 " Hint: the backend rejected the Bearer token. Ensure you "
257 "signed in successfully."
258 )
259 _fail( f " { err }{ hint } " )
260 except Exception :
261 # Log the full exception locally; return a generic message.
262 logger.debug( "Unhandled error in command %s " , command, exc_info = True )
263 _fail( "The operation failed. Re-run with logging enabled for details." )
264
265
266 if __name__ == "__main__" :
267 main()