|
1 | 1 | import asyncio |
| 2 | +import subprocess |
2 | 3 |
|
3 | 4 | import httpx |
4 | 5 | import turbopuffer as tpuf |
@@ -225,3 +226,69 @@ def display_callable_signature(import_path: str) -> str: |
225 | 226 | >>> display_callable_signature('prefect:flow') |
226 | 227 | """ |
227 | 228 | return display_signature(import_path) |
| 229 | + |
| 230 | + |
| 231 | +def check_cli_command(command: str, args: list[str] | None = None) -> str: |
| 232 | + """ |
| 233 | + Run a CLI command to verify its behavior or check help documentation. |
| 234 | +
|
| 235 | + This tool is specifically designed to help verify Prefect CLI commands before suggesting them to users. |
| 236 | + Use this to check if a command exists, what its options are, or to verify the correct syntax. |
| 237 | +
|
| 238 | + IMPORTANT USAGE GUIDELINES: |
| 239 | + - Use this tool BEFORE suggesting any CLI command to a user |
| 240 | + - Always check with --help first to verify command structure |
| 241 | + - Common commands to verify: prefect deploy, prefect work-pool, prefect worker, etc. |
| 242 | + - This helps prevent suggesting non-existent or incorrectly formatted commands |
| 243 | +
|
| 244 | + Args: |
| 245 | + command: The base command to run (e.g., "prefect", "prefect deploy") |
| 246 | + args: Additional arguments to pass (e.g., ["--help"], ["work-pool", "create", "--help"]) |
| 247 | +
|
| 248 | + Returns: |
| 249 | + The output of the command or an error message if it fails |
| 250 | +
|
| 251 | + Examples: |
| 252 | + # Check if a command exists and see its options |
| 253 | + >>> check_cli_command("prefect deploy", ["--help"]) |
| 254 | +
|
| 255 | + # Verify work pool commands |
| 256 | + >>> check_cli_command("prefect work-pool", ["--help"]) |
| 257 | +
|
| 258 | + # Check specific subcommand help |
| 259 | + >>> check_cli_command("prefect", ["worker", "start", "--help"]) |
| 260 | + """ |
| 261 | + if args is None: |
| 262 | + args = [] |
| 263 | + |
| 264 | + # Construct the full command |
| 265 | + full_command = command.split() + args |
| 266 | + |
| 267 | + try: |
| 268 | + # Run the command with a timeout |
| 269 | + result = subprocess.run( |
| 270 | + full_command, |
| 271 | + capture_output=True, |
| 272 | + text=True, |
| 273 | + timeout=10, |
| 274 | + check=False, # Don't raise on non-zero exit codes |
| 275 | + ) |
| 276 | + |
| 277 | + # Combine stdout and stderr for complete output |
| 278 | + output = "" |
| 279 | + if result.stdout: |
| 280 | + output += result.stdout |
| 281 | + if result.stderr: |
| 282 | + output += f"\n[stderr]: {result.stderr}" |
| 283 | + |
| 284 | + if not output.strip(): |
| 285 | + output = f"Command ran successfully with exit code {result.returncode} but produced no output" |
| 286 | + |
| 287 | + return output[:2000] # Limit output length to prevent huge responses |
| 288 | + |
| 289 | + except subprocess.TimeoutExpired: |
| 290 | + return "Command timed out after 10 seconds" |
| 291 | + except FileNotFoundError: |
| 292 | + return f"Command '{full_command[0]}' not found. Make sure it's installed and in PATH." |
| 293 | + except Exception as e: |
| 294 | + return f"Error running command: {str(e)}" |
0 commit comments