Got it! You can trigger a Prefect 3.x deployment via the REST API using the deployment’s flow name and deployment name, and pass parameters with a simple POST request. Key points: - Endpoint: POST {PREFECT_API_URL}/deployments/name/{flow_name}/{deployment_name}/create_flow_run - Body: JSON with

parameters

- Auth: Include Authorization: Bearer <your token>. Since you enabled

PREFECT_SERVER_API_AUTH_STRING

, your request must include this header. Note that

PREFECT_API_AUTH_STRING

only affects Prefect’s Python client—it won’t be applied automatically when you use

requests

. Example using python-requests (URL-encoding names is important if they contain spaces or special characters):

import os
import requests
from urllib.parse import quote

PREFECT_API_URL = os.environ.get("PREFECT_API_URL", "<http://localhost:4200/api>")

# If you set PREFECT_API_AUTH_STRING on this process, you can read it here
API_TOKEN = os.environ["PREFECT_API_AUTH_STRING"]  # or set manually as a string

flow_name = "YOUR_FLOW_NAME"
deployment_name = "YOUR_DEPLOYMENT_NAME"

endpoint = f"{PREFECT_API_URL}/deployments/name/{quote(flow_name, safe='')}/{quote(deployment_name, safe='')}/create_flow_run"

payload = {
    "parameters": {
        "seller_id": "xx",
        "selling_partner_id": "x1",
        "data_cycle": "t1"
    }
    # Optional fields you can add:
    # "idempotency_key": "unique-key-here",
    # "tags": ["triggered-via-api"],
    # "state": {"type": "SCHEDULED"}  # or "PENDING" to start ASAP
}

headers = {
    "Authorization": f"Bearer {API_TOKEN}",
    "Content-Type": "application/json",
}

resp = <http://requests.post|requests.post>(endpoint, json=payload, headers=headers, timeout=30)
resp.raise_for_status()
flow_run = resp.json()
print("Created flow run:", flow_run["id"])

Optional: poll the flow run until it reaches a terminal state

flow_run_id = flow_run["id"]
status_url = f"{PREFECT_API_URL}/flow_runs/{flow_run_id}"

while True:
    r = requests.get(status_url, headers=headers, timeout=15)
    r.raise_for_status()
    state = r.json()["state"]["type"]
    print("State:", state)
    if state in {"COMPLETED", "FAILED", "CANCELLED", "CRASHED"}:
        break

Notes: - If you prefer to use the deployment ID instead, the parallel endpoint is POST {PREFECT_API_URL}/deployments/{deployment_id}/create_flow_run with the same payload. - If you get 401/403 errors, make sure the token in your Authorization header exactly matches the server’s configured

PREFECT_SERVER_API_AUTH_STRING

. - The

requests

library won’t pick up

PREFECT_API_AUTH_STRING

automatically; you must pass it explicitly as shown. If you share your flow name and deployment name (or any error you hit), I can tailor the exact call for your environment.

You’re hitting two issues: 1) Auth header type is wrong for a self-hosted Server -

PREFECT_API_KEY

(Bearer token) is for Prefect Cloud. For self‑hosted Prefect Server with auth enabled, you must use HTTP Basic auth derived from

PREFECT_API_AUTH_STRING

(format

username:password

). - On the server, set

PREFECT_SERVER_API_AUTH_STRING=cldsadminn:123

. On the client, use

PREFECT_API_AUTH_STRING=cldsadminn:123

or pass the same via Basic auth headers. - Do not send

Authorization: Bearer ...

to a self-hosted Server; send

Authorization: Basic <base64(username:password)>

or use `requests`’ built-in

auth=(user, pass)

. 2) Endpoint path is off - In Prefect 3.x, there isn’t a POST at

/deployments/name/{flow}/{deployment}/create_flow_run

. - The correct flow is: - GET

/api/deployments/name/{flow_name}/{deployment_name}

to get the deployment ID - POST

/api/deployments/{deployment_id}/flow_runs

with your parameters Working example with

requests

and Basic auth:

import requests
from urllib.parse import quote

PREFECT_API_URL = "<http://localhost:4200/api>"
USERNAME = "cldsadminn"
PASSWORD = "123"

def get_deployment_id(flow_name: str, deployment_name: str) -> str:
    url = f"{PREFECT_API_URL}/deployments/name/{quote(flow_name)}/{quote(deployment_name)}"
    r = requests.get(url, auth=(USERNAME, PASSWORD))
    r.raise_for_status()
    data = r.json()
    # API returns the full deployment object; id is under 'id' or 'deployment_id' depending on version
    return data.get("id") or data["deployment_id"]

def create_flow_run_from_deployment(deployment_id: str, parameters: dict) -> dict:
    url = f"{PREFECT_API_URL}/deployments/{deployment_id}/flow_runs"
    payload = {"parameters": parameters}
    r = <http://requests.post|requests.post>(url, json=payload, auth=(USERNAME, PASSWORD))
    r.raise_for_status()
    return r.json()

def run_deployment_by_full_name(flow_name, deployment_name, parameters):
    dep_id = get_deployment_id(flow_name, deployment_name)
    return create_flow_run_from_deployment(dep_id, parameters)

FLOW_NAME = "EBAY"
DEPLOYMENT_NAME = "dev"
PARAMETERS = {
    "seller_id": "d480fd76-f59e-4ea0-95cb-4e22db0164e0",
    "selling_partner_id": "ed442813-afe7-4026-bc14-190ca41bfb23",
    "data_cycle": "origination",
}

flow_run = run_deployment_by_full_name(FLOW_NAME, DEPLOYMENT_NAME, PARAMETERS)
print(flow_run)

If you prefer to keep constructing the header yourself:

import base64
token = base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode()
headers = {"Authorization": f"Basic {token}", "Content-Type": "application/json"}

Key checks if you still see 401: - Confirm the server really has

PREFECT_SERVER_API_AUTH_STRING=cldsadminn:123

set in the environment where

prefect server start

is running, and that there’s no extra whitespace. - Ensure your client is not setting

PREFECT_API_KEY

(Cloud-only). If both are set in a Python SDK client,

PREFECT_API_KEY

takes precedence and will fail against Server. - Verify you’re calling the right host.

<http://localhost:4200>

must point to the machine running Prefect Server from the code’s point of view. Docs for reference: - Prefect 3 Server auth settings: Security settings - Read deployment by name (GET): API: read_deployment_by_name - Create flow run from deployment (POST): API: create_flow_run_from_deployment Do you want a version of this using the Prefect Python client instead of raw HTTP? It simplifies auth and endpoint details.