Let me show you how to create and submit a flow run through Prefect's API. There are two main approaches: 1. Using the Python Client:

python
from prefect.client import get_client
from prefect import flow

async def create_flow_run():
    # Get the async client
    async with get_client() as client:
        # Create a flow run
        flow_run = await client.create_flow_run(
            flow_name="your-flow-name",
            parameters={"param1": "value1"}  # Optional parameters
        )
        print(f"Created flow run with ID: {flow_run.id}")

For synchronous usage:

python
from prefect.client import sync_get_client

# Get the sync client
client = sync_get_client()

# Create a flow run
flow_run = client.create_flow_run(
    flow_name="your-flow-name",
    parameters={"param1": "value1"}  # Optional parameters
)
print(f"Created flow run with ID: {flow_run.id}")

2. Using Direct REST API Calls: If you prefer using HTTP directly, you can use the REST API endpoint. Here's an example using `curl`:

bash
curl -X POST "<http://your-prefect-server/api/flow_runs/create>" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -d '{
    "flow_name": "your-flow-name",
    "parameters": {
      "param1": "value1"
    }
  }'

Key Points: - You'll need to have your Prefect server running and accessible - Make sure you have the proper authentication (API key if required) - The flow must be registered with the server before you can create runs - You can specify parameters if your flow accepts them - The response will include the flow run ID which you can use to track the run's status For more complex scenarios, you can also specify additional parameters like: - `deployment_name`: If you want to run a specific deployment - `state`: To set an initial state for the flow run - `tags`: To add tags to the flow run Remember that if you're using the API in a production environment, it's recommended to use API keys for authentication. You can set this up in your Prefect server configuration.

Here's a breakdown of the essential environment variables for the Prefect client: For Prefect Cloud:

bash
PREFECT_API_URL=<https://api.prefect.cloud/api/accounts/{account-id}/workspaces/{workspace-id}>
PREFECT_API_KEY=pnu_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

For Self-hosted Prefect Server:

bash
PREFECT_API_URL=<http://your-server-address:4200/api>
PREFECT_API_AUTH_STRING=admin:your-password  # Optional, only if you've configured authentication

Key Points: -

PREFECT_API_URL

is always required and must point to your server's API endpoint - For Prefect Cloud: -

PREFECT_API_KEY

is required - Do not set

PREFECT_API_AUTH_STRING

- For self-hosted servers: -

PREFECT_API_AUTH_STRING

is optional and depends on your server's authentication configuration - Do not set

PREFECT_API_KEY

If these environment variables are properly set, the client will automatically pick them up when initialized:

python
from prefect.client import get_client

# The client will automatically use the environment variables
async with get_client() as client:
    # Your client operations here
    pass

You can verify your connection by attempting a simple operation:

python
async with get_client() as client:
    # This will fail if connection isn't properly configured
    workspaces = await client.read_workspaces()

Remember that these environment variables can be set in various ways: 1. In your shell environment 2. In a

.env

file in your project directory 3. In your deployment configuration 4. In your container environment if using containerized workflows For security best practices, never hardcode these values in your code, especially the API key or authentication string.