Coverage for src / osiris_cli / cloud_tools_letta.py: 0%

1""" 

2Cloud Tools with Letta Visual Standard 

3 

4AI-friendly cloud tools using Letta formatting. 

5""" 

6 

7from typing import Dict, Any, Optional 

8 

9from .cloud_manager import cloud_manager, CloudProvider 

10from .letta_formatter import ( 

11 action_result, 

12 thinking, 

13 LettaGlyphs 

14) 

15from .logger import get_logger 

16from .errors import CloudError 

17 

18logger = get_logger() 

19 

20 

21# Tool definitions for AI 

22CLOUD_TOOLS_LETTA = [ 

23 { 

24 "type": "function", 

25 "function": { 

26 "name": "aws_list_ec2", 

27 "description": "List AWS EC2 instances in a region. Shows instance IDs, types, states, and IPs.", 

28 "parameters": { 

29 "type": "object", 

30 "properties": { 

31 "region": { 

32 "type": "string", 

33 "description": "AWS region (default: us-east-1)", 

34 "default": "us-east-1" 

35 } 

36 } 

37 } 

38 } 

39 }, 

40 { 

41 "type": "function", 

42 "function": { 

43 "name": "aws_list_s3", 

44 "description": "List all AWS S3 buckets in the account.", 

45 "parameters": { 

46 "type": "object", 

47 "properties": {} 

48 } 

49 } 

50 }, 

51 { 

52 "type": "function", 

53 "function": { 

54 "name": "aws_list_lambda", 

55 "description": "List AWS Lambda functions in a region.", 

56 "parameters": { 

57 "type": "object", 

58 "properties": { 

59 "region": { 

60 "type": "string", 

61 "description": "AWS region (default: us-east-1)", 

62 "default": "us-east-1" 

63 } 

64 } 

65 } 

66 } 

67 }, 

68 { 

69 "type": "function", 

70 "function": { 

71 "name": "gcp_list_compute", 

72 "description": "List GCP Compute Engine instances in a project and zone.", 

73 "parameters": { 

74 "type": "object", 

75 "properties": { 

76 "project": { 

77 "type": "string", 

78 "description": "GCP project ID" 

79 }, 

80 "zone": { 

81 "type": "string", 

82 "description": "GCP zone (default: us-central1-a)", 

83 "default": "us-central1-a" 

84 } 

85 }, 

86 "required": ["project"] 

87 } 

88 } 

89 }, 

90 { 

91 "type": "function", 

92 "function": { 

93 "name": "azure_list_vms", 

94 "description": "List Azure Virtual Machines, optionally filtered by resource group.", 

95 "parameters": { 

96 "type": "object", 

97 "properties": { 

98 "resource_group": { 

99 "type": "string", 

100 "description": "Azure resource group (optional)" 

101 } 

102 } 

103 } 

104 } 

105 }, 

106 { 

107 "type": "function", 

108 "function": { 

109 "name": "cloud_status", 

110 "description": "Check authentication status and CLI availability for all cloud providers.", 

111 "parameters": { 

112 "type": "object", 

113 "properties": {} 

114 } 

115 } 

116 } 

117] 

118 

119 

120def summarize_instances(result: Dict[str, Any]) -> str: 

121 """Summarize instance list results""" 

122 count = result.get("count", 0) 

123 provider = result.get("provider", "cloud") 

124 resource_type = result.get("resource_type", "resources") 

125 

126 if count == 0: 

127 return f"No {resource_type} found in {provider}" 

128 

129 # Get status breakdown if available 

130 if "instances" in result: 

131 instances = result["instances"] 

132 if instances and "state" in instances[0]: 

133 states = {} 

134 for inst in instances: 

135 state = inst.get("state", "unknown") 

136 states[state] = states.get(state, 0) + 1 

137 

138 state_str = ", ".join(f"{count} {state}" for state, count in states.items()) 

139 return f"Found {count} instances: {state_str}" 

140 

141 return f"Found {count} {resource_type}" 

142 

143 

144# Tool implementations with Letta formatting 

145def tool_aws_list_ec2_letta(region: str = "us-east-1") -> Dict[str, Any]: 

146 """List AWS EC2 instances with Letta formatting""" 

147 

148 display_args = {"region": region} 

149 

150 try: 

151 result = cloud_manager.aws_list_ec2(region) 

152 

153 summary = summarize_instances(result) 

154 

155 action_result( 

156 "aws_list_ec2", 

157 display_args, 

158 summary, 

159 success=True 

160 ) 

161 

162 return result 

163 

164 except Exception as e: 

165 logger.error(f"AWS EC2 list failed: {e}") 

166 

167 error_msg = str(e) 

168 action_result( 

169 "aws_list_ec2", 

170 display_args, 

171 f"Failed: {error_msg}", 

172 success=False 

173 ) 

174 

175 return { 

176 "success": False, 

177 "error": error_msg 

178 } 

179 

180 

181def tool_aws_list_s3_letta() -> Dict[str, Any]: 

182 """List AWS S3 buckets with Letta formatting""" 

183 

184 display_args = {} 

185 

186 try: 

187 result = cloud_manager.aws_list_s3() 

188 

189 count = result.get("count", 0) 

190 summary = f"Found {count} S3 buckets" if count > 0 else "No S3 buckets found" 

191 

192 action_result( 

193 "aws_list_s3", 

194 display_args, 

195 summary, 

196 success=True 

197 ) 

198 

199 return result 

200 

201 except Exception as e: 

202 logger.error(f"AWS S3 list failed: {e}") 

203 

204 error_msg = str(e) 

205 action_result( 

206 "aws_list_s3", 

207 display_args, 

208 f"Failed: {error_msg}", 

209 success=False 

210 ) 

211 

212 return { 

213 "success": False, 

214 "error": error_msg 

215 } 

216 

217 

218def tool_aws_list_lambda_letta(region: str = "us-east-1") -> Dict[str, Any]: 

219 """List AWS Lambda functions with Letta formatting""" 

220 

221 display_args = {"region": region} 

222 

223 try: 

224 result = cloud_manager.aws_list_lambda(region) 

225 

226 count = result.get("count", 0) 

227 summary = f"Found {count} Lambda functions" if count > 0 else "No Lambda functions found" 

228 

229 action_result( 

230 "aws_list_lambda", 

231 display_args, 

232 summary, 

233 success=True 

234 ) 

235 

236 return result 

237 

238 except Exception as e: 

239 logger.error(f"AWS Lambda list failed: {e}") 

240 

241 error_msg = str(e) 

242 action_result( 

243 "aws_list_lambda", 

244 display_args, 

245 f"Failed: {error_msg}", 

246 success=False 

247 ) 

248 

249 return { 

250 "success": False, 

251 "error": error_msg 

252 } 

253 

254 

255def tool_gcp_list_compute_letta(project: str, zone: str = "us-central1-a") -> Dict[str, Any]: 

256 """List GCP Compute instances with Letta formatting""" 

257 

258 display_args = {"project": project, "zone": zone} 

259 

260 try: 

261 result = cloud_manager.gcp_list_compute(project, zone) 

262 

263 summary = summarize_instances(result) 

264 

265 action_result( 

266 "gcp_list_compute", 

267 display_args, 

268 summary, 

269 success=True 

270 ) 

271 

272 return result 

273 

274 except Exception as e: 

275 logger.error(f"GCP Compute list failed: {e}") 

276 

277 error_msg = str(e) 

278 action_result( 

279 "gcp_list_compute", 

280 display_args, 

281 f"Failed: {error_msg}", 

282 success=False 

283 ) 

284 

285 return { 

286 "success": False, 

287 "error": error_msg 

288 } 

289 

290 

291def tool_azure_list_vms_letta(resource_group: Optional[str] = None) -> Dict[str, Any]: 

292 """List Azure VMs with Letta formatting""" 

293 

294 display_args = {} 

295 if resource_group: 

296 display_args["resource_group"] = resource_group 

297 

298 try: 

299 result = cloud_manager.azure_list_vms(resource_group) 

300 

301 summary = summarize_instances(result) 

302 

303 action_result( 

304 "azure_list_vms", 

305 display_args, 

306 summary, 

307 success=True 

308 ) 

309 

310 return result 

311 

312 except Exception as e: 

313 logger.error(f"Azure VM list failed: {e}") 

314 

315 error_msg = str(e) 

316 action_result( 

317 "azure_list_vms", 

318 display_args, 

319 f"Failed: {error_msg}", 

320 success=False 

321 ) 

322 

323 return { 

324 "success": False, 

325 "error": error_msg 

326 } 

327 

328 

329def tool_cloud_status_letta() -> Dict[str, Any]: 

330 """Check cloud provider status with Letta formatting""" 

331 

332 display_args = {} 

333 

334 try: 

335 status = cloud_manager.get_provider_status() 

336 

337 # Count available providers 

338 available = sum(1 for p, s in status.items() if s["cli_available"]) 

339 authenticated = sum(1 for p, s in status.items() if s["authenticated"]) 

340 

341 summary = f"Available: {available}/3 providers, Authenticated: {authenticated}/3" 

342 

343 action_result( 

344 "cloud_status", 

345 display_args, 

346 summary, 

347 success=True 

348 ) 

349 

350 return { 

351 "success": True, 

352 "status": status, 

353 "summary": summary 

354 } 

355 

356 except Exception as e: 

357 logger.error(f"Cloud status check failed: {e}") 

358 

359 error_msg = str(e) 

360 action_result( 

361 "cloud_status", 

362 display_args, 

363 f"Failed: {error_msg}", 

364 success=False 

365 ) 

366 

367 return { 

368 "success": False, 

369 "error": error_msg 

370 } 

371 

372 

373# Tool registry 

374TOOL_HANDLERS_CLOUD_LETTA = { 

375 "aws_list_ec2": tool_aws_list_ec2_letta, 

376 "aws_list_s3": tool_aws_list_s3_letta, 

377 "aws_list_lambda": tool_aws_list_lambda_letta, 

378 "gcp_list_compute": tool_gcp_list_compute_letta, 

379 "azure_list_vms": tool_azure_list_vms_letta, 

380 "cloud_status": tool_cloud_status_letta 

381} 

382 

383 

384# Example workflow 

385def example_cloud_workflow(): 

386 """ 

387 Example cloud workflow with Letta formatting. 

388  

389 Output: 

390 ✻ Thinking... 

391 I'll check cloud provider status first, then list AWS resources. 

392  

393 ● cloud_status() 

394 ⎿ Available: 3/3 providers, Authenticated: 2/3 ✓ 

395  

396 ● aws_list_ec2(region="us-east-1") 

397 ⎿ Found 3 instances: 2 running, 1 stopped ✓ 

398  

399 ● aws_list_s3() 

400 ⎿ Found 5 S3 buckets ✓ 

401 """ 

402 

403 thinking("I'll check cloud provider status first to see what's available, then list AWS resources.") 

404 

405 # Check status 

406 tool_cloud_status_letta() 

407 

408 # List AWS resources 

409 tool_aws_list_ec2_letta("us-east-1") 

410 tool_aws_list_s3_letta() 

411 

412 

413if __name__ == "__main__": 

414 print("\n" + "="*60) 

415 print("LETTA VISUAL STANDARD - CLOUD OPERATIONS") 

416 print("="*60 + "\n") 

417 

418 example_cloud_workflow() 

419 

420 print("\n" + "="*60 + "\n")