Here's a complete example of how to trigger a flow run via REST API with custom parameters: 1. First, let's assume you have a deployed flow. Here's a simple example flow with parameters:

python
from prefect import flow

@flow
def my_parameterized_flow(name: str, count: int = 1):
    print(f"Hello {name}, {count} times!")

2. To trigger this flow via REST API, you'll need to make a POST request to the Prefect API. Here's how to do it using Python's requests library:

python
import requests

# Your Prefect API URL (default for local server)
api_url = "<http://127.0.0.1:4200/api>"

# Your API key (if using Prefect Cloud)
api_key = "your-api-key"  # Only needed for Prefect Cloud

# Headers
headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {api_key}"  # Only needed for Prefect Cloud
}

# Request body
payload = {
    "deployment_id": "your-deployment-id",  # Replace with your deployment ID
    "parameters": {
        "name": "Alice",
        "count": 3
    }
}

# Make the API request
response = <http://requests.post|requests.post>(
    f"{api_url}/flow_runs/",
    json=payload,
    headers=headers
)

print(response.json())

If you prefer using curl, here's the equivalent command:

bash
# For local Prefect server
curl -X POST "<http://127.0.0.1:4200/api/flow_runs/>" \
     -H "Content-Type: application/json" \
     -d '{
           "deployment_id": "your-deployment-id",
           "parameters": {
             "name": "Alice",
             "count": 3
           }
         }'

# For Prefect Cloud (include API key)
curl -X POST "<https://api.prefect.cloud/api/accounts/{account-id}/workspaces/{workspace-id}/flow_runs/>" \
     -H "Authorization: Bearer your-api-key" \
     -H "Content-Type: application/json" \
     -d '{
           "deployment_id": "your-deployment-id",
           "parameters": {
             "name": "Alice",
             "count": 3
           }
         }'

Key points: - Replace

your-deployment-id

with your actual deployment ID - For Prefect Cloud, you'll need to include your API key in the Authorization header - The

parameters

object in the payload should match the parameter names in your flow - Additional optional fields like

tags

,

state

, or

job_variables

can be included in the payload You can find the complete API reference documentation at: https://docs-3.prefect.io/api-ref/rest-api/server/flow-runs/create-flow-run

Let me update the previous example with authentication and immediate execution details: 1. First, set up authentication on your Prefect server. In your server configuration (typically

~/.prefect/server.yaml

):

yaml
server:
  api:
    auth_string: "admin:your-password"  # Replace with your desired credentials

2. On the client side, set the same authentication credentials. You can do this in several ways: - Environment variable:

bash
     export PREFECT_SERVER_API_AUTH_STRING="admin:your-password"

- In your Python code:

python
     from prefect.settings import set_setting
     set_setting("PREFECT_SERVER_API_AUTH_STRING", "admin:your-password")

3. Here's the updated API call with authentication and immediate execution:

python
import requests

# Your Prefect API URL
api_url = "<http://127.0.0.1:4200/api>"

# Basic auth credentials
import base64
auth_string = "admin:your-password"
encoded_auth = base64.b64encode(auth_string.encode()).decode()

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Basic {encoded_auth}"
}

payload = {
    "deployment_id": "your-deployment-id",
    "state": {
        "type": "PENDING",  # This triggers immediate execution
        "name": "Pending"
    },
    "parameters": {
        "name": "Alice",
        "count": 3
    }
}

response = <http://requests.post|requests.post>(
    f"{api_url}/flow_runs/",
    json=payload,
    headers=headers
)

print(response.json())

To specify which worker should run your flow: 1. The worker selection is determined by the work pool specified in your deployment. When you create a deployment, you specify a work pool, which determines the type of worker that can run it:

python
from prefect import flow
from prefect.deployments import Deployment

@flow
def my_flow(name: str, count: int = 1):
    print(f"Hello {name}, {count} times!")

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="my-deployment",
    work_pool_name="my-process-pool",  # This determines which worker picks up the work
    work_queue_name="default"          # Optional: for more granular control
)

2. When starting your worker, make sure it's configured to watch the correct work pool:

bash
prefect worker start -p "my-process-pool"

Key points: - For self-hosted Prefect, use Basic Authentication with

auth_string

- Set state to

"PENDING"

for immediate execution - Worker selection is controlled by the work pool specified in your deployment - Work pools can be of different types (process, docker, kubernetes) - Multiple workers can watch the same work pool for high availability - Work queues within work pools can provide more granular control over which worker picks up specific work Remember that workers must be running and healthy to pick up work from their assigned work pools. You can monitor worker health in the Prefect UI or via the API.