Great question. Here’s a clean, secure way to do this from a Next.js app with Prefect 3.x: High-level - Best practice: create your deployment once (via code/CLI/CI), then from your app just create flow runs with parameters and show their status. - If you truly need to create deployments via API at runtime, it’s possible but more complex (you must provide flow_id, entrypoint, work pool, etc.). I’ve included pointers below. Security note: never expose your Prefect API key in the browser. Do API calls to Prefect from Next.js server code (API routes or server actions), then have the client call your API. Setup - Prefect Cloud API base: - API: https://api.prefect.cloud/api/accounts/{ACCOUNT_ID}/workspaces/{WORKSPACE_ID} - UI: https://app.prefect.cloud/account/{ACCOUNT_ID}/workspace/{WORKSPACE_ID} - Auth header: Authorization: Bearer {PREFECT_API_KEY} - Self-hosted API base (example): http://localhost:4200/api Trigger a run with parameters (recommended path) 1) Server route to create a flow run from a deployment by name Create a Next.js API route that: - Looks up the deployment by name - Creates a flow run with input parameters - Returns the flow_run_id and a link to your Prefect UI

/**
 * POST /api/prefect/run
 * Body: { flowName: string, deploymentName: string, parameters?: object, runName?: string, tags?: string[] }
 */
export default async function handler(req, res) {
  if (req.method !== 'POST') return res.status(405).end();

  const {
    flowName,
    deploymentName,
    parameters = {},
    runName,
    tags = []
  } = req.body || {};

  if (!flowName || !deploymentName) {
    return res.status(400).json({ error: 'flowName and deploymentName are required' });
  }

  const ACCOUNT_ID = process.env.PREFECT_ACCOUNT_ID;
  const WORKSPACE_ID = process.env.PREFECT_WORKSPACE_ID;
  const API_KEY = process.env.PREFECT_API_KEY;
  const BASE = `<https://api.prefect.cloud/api/accounts/${ACCOUNT_ID}/workspaces/${WORKSPACE_ID}`;>

  try {
    // 1) Lookup deployment by name
    const depResp = await fetch(
      `${BASE}/deployments/name/${encodeURIComponent(flowName)}/${encodeURIComponent(deploymentName)}`,
      {
        headers: {
          'Authorization': `Bearer ${API_KEY}`
        }
      }
    );

    if (!depResp.ok) {
      const err = await depResp.text();
      return res.status(depResp.status).json({ error: `Deployment lookup failed: ${err}` });
    }
    const deployment = await depResp.json();

    // 2) Create flow run from deployment with parameters
    const runResp = await fetch(
      `${BASE}/deployments/${deployment.id}/create_flow_run`,
      {
        method: 'POST',
        headers: {
          'Authorization': `Bearer ${API_KEY}`,
          'Content-Type': 'application/json'
        },
        body: JSON.stringify({
          name: runName,          // optional
          parameters,             // your input parameters
          tags,                   // optional
          // idempotency_key: 'optional-string'  // optional but useful if you want deduplication
        })
      }
    );

    if (!runResp.ok) {
      const err = await runResp.text();
      return res.status(runResp.status).json({ error: `Flow run creation failed: ${err}` });
    }

    const flowRun = await runResp.json();
    const uiUrl = `<https://app.prefect.cloud/account/${ACCOUNT_ID}/workspace/${WORKSPACE_ID}/flow-runs/flow-run/${flowRun.id}`;>

    return res.status(200).json({
      flow_run_id: flowRun.id,
      state: flowRun.state?.type,
      ui_url: uiUrl
    });
  } catch (e) {
    return res.status(500).json({ error: e.message || String(e) });
  }
}

2) Server route to poll flow run status ``` /** * GET /api/prefect/run/[id] */ export default async function handler(req, res) { const { id } = req.query; if (!id) return res.status(400).json({ error: 'Missing flow run id' }); const ACCOUNT_ID = process.env.PREFECT_ACCOUNT_ID; const WORKSPACE_ID = process.env.PREFECT_WORKSPACE_ID;

const API_KEY = process.env.PREFECT_API_KEY; const BASE = `https://api.prefect.cloud/api/accounts/${ACCOUNT_ID}/workspaces/${WORKSPACE_ID}`; try { const resp = await fetch(

${BASE}/flow_runs/${id}

, { headers: { 'Authorization':

Bearer ${API_KEY}

} }); if (!resp.ok) { const err = await resp.text(); return res.status(resp.status).json({ error: err }); } const fr = await resp.json(); return res.status(200).json({ id: fr.id, state_type: fr.state?.type, // SCHEDULED, RUNNING, COMPLETED, FAILED, CRASHED state_name: fr.state?.name, updated: fr.updated }); } catch (e) { return res.status(500).json({ error: e.message || String(e) }); } }

3) Client usage example

- POST to your /api/prefect/run with parameters
- Then poll /api/prefect/run/[id] every few seconds until state_type is terminal (COMPLETED/FAILED/CRASHED)

async function startRun() { const res = await fetch('/api/prefect/run', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ flowName: 'my-flow', deploymentName: 'prod', parameters: { foo: 'bar', n: 3 }, runName: 'run-from-nextjs' }) }); const { flow_run_id, ui_url, state, error } = await res.json(); if (error) throw new Error(error); // Show link to Prefect UI immediately console.log('Run URL:', ui_url); // Poll status const interval = setInterval(async () => { const sres = await fetch(

/api/prefect/run/${flow_run_id}

); const s = await sres.json(); console.log('State:', s.state_type); if (['COMPLETED', 'FAILED', 'CRASHED'].includes(s.state_type)) { clearInterval(interval); // Update your UI accordingly } }, 3000); } ``` If you truly want to create deployments via API at runtime - It’s possible, but you must supply the full deployment configuration, including flow_id, entrypoint/path or pull steps (code location), work_pool_name, etc. This is usually better handled in CI or as code with flow.from_source(...).deploy(...). - REST endpoints: - Create deployment: POST /deployments - Read deployment by name: GET /deployments/name/{flow_name}/{deployment_name} - Create flow run from deployment: POST /deployments/{deployment_id}/create_flow_run - Docs: - Create flow run from deployment: REST: create flow run from deployment - Read deployment by name: REST: read deployment by name - Create deployment: REST: create deployment - Read flow run: REST: read flow run - REST API index: Prefect 3.x REST API Notes and tips - Parameters: pass as JSON under

parameters

in the create_flow_run request. They must match your flow’s parameter schema. - Idempotency: you can include

idempotency_key

in the create_flow_run body to avoid duplicate runs if your user double-clicks. - UI deep link: for Cloud, it’s https://app.prefect.cloud/account/{ACCOUNT_ID}/workspace/{WORKSPACE_ID}/flow-runs/flow-run/{flow_run_id}. - Keep secrets on the server: store PREFECT_API_KEY, ACCOUNT_ID, WORKSPACE_ID as env vars and only call Prefect from server-side code. Questions for you - Are you on Prefect Cloud or self-hosted? - Do you want to create deployments dynamically from the app, or is triggering runs of an existing deployment sufficient? If you share your desired shape of parameters, I can tailor the request body and validation.