prefect-client

import os import asyncio from prefect.deployments import run_deployment from prefect.client.orchestration import get_client # Ensure API URL and Key are set as environment variables for remote connection os.environ['PREFECT_API_URL'] = os.environ.get('PREFECT_API_URL', 'http://localhost:4200/api') os.environ['PREFECT_API_KEY'] = os.environ.get('PREFECT_API_KEY', 'your_api_key_if_needed') async def interact_with_prefect_api(): print(f"Connecting to Prefect API at: {os.environ['PREFECT_API_URL']}") async with get_client() as client: # Example 1: Remotely trigger a deployment try: flow_run = await run_deployment( name="my-flow/my-deployment", parameters={"message": "Hello from prefect-client!"}, timeout=0 # Do not wait for the flow run to complete ) print(f"Triggered flow run: {flow_run.name} (ID: {flow_run.id})") except Exception as e: print(f"Could not trigger deployment: {e}. Ensure 'my-flow/my-deployment' exists and is accessible.") # Example 2: Read some information from the API try: deployments = await client.read_deployments(limit=5) print(f"\nFound {len(deployments)} deployments:") for dep in deployments: print(f" - {dep.name} (ID: {dep.id})") except Exception as e: print(f"Could not read deployments: {e}. Check API URL/Key and server status.") if __name__ == "__main__": # Note: If connecting to Prefect Cloud, ensure PREFECT_API_KEY is set # os.environ['PREFECT_API_KEY'] = 'your_prefect_cloud_api_key' asyncio.run(interact_with_prefect_api())