Great question. In Prefect 3.x, you have two solid patterns for getting a deployment’s result back to the caller (your LLM tool): 1) Synchronous request/response (LLM waits for the run to finish) - Trigger the deployment and block until it completes, then return the flow’s return value.

from prefect.deployments import run_deployment

def run_and_return(deployment_name: str, params: dict, timeout_seconds: int = 300):
    # waits by default; set timeout to bound how long you’ll wait (None = wait indefinitely, 0 = return immediately)
    flow_run = run_deployment(
        name=deployment_name,               # "flow-name/deployment-name"
        parameters=params,
        timeout=timeout_seconds,            # e.g., 300 seconds
        poll_interval=2                     # optional
    )

    # Get the returned value from the flow
    # Raises on failure by default; you can control behavior with raise_on_failure=False
    result = flow_run.state.result()

    return {
        "flow_run_id": str(flow_run.id),
        "status": "COMPLETED" if flow_run.state.is_completed() else flow_run.state.type.value,
        "result": result
    }

Notes: - This returns only after the run reaches a final state or your timeout hits. - The returned value should be JSON-serializable if your LLM tool expects JSON. - Prefect 3.x persists results by default; you can also write your own artifacts/links for larger outputs (see below). 2) Async callback (LLM does not wait; you notify it later) - Start the flow run and return immediately with a correlation ID. - Use an Automation to POST to your webhook when the flow completes (include metadata + links/artifacts), or poll/wait in your own service. Start the run and return a correlation ID:

from prefect.deployments import run_deployment

def start_and_return_id(deployment_name: str, params: dict):
    flow_run = run_deployment(
        name=deployment_name,
        parameters=params,
        timeout=0  # return immediately
    )
    return {"flow_run_id": str(flow_run.id), "status": "SCHEDULED"}

Then configure an Automation to call your webhook on completion: - Create a Webhook Block (contains URL/headers) - Create an Automation with a call-webhook action that triggers on flow-run completion Example Automation (YAML):

name: Notify on Flow Completion
description: Send a webhook when a flow run completes
enabled: true
trigger:
  type: event
  posture: Reactive
  expect:
    - prefect.flow-run.Completed
  threshold: 1
actions:
  - type: call-webhook
    block_document_id: "YOUR_WEBHOOK_BLOCK_ID"
    payload: |
      {
        "event": "flow_completed",
        "flow_run_id": "{{ flow_run.id }}",
        "flow_run_name": "{{ flow_run.name }}",
        "status": "{{ flow_run.state.name }}",
        "ui_url": "{{ flow_run|ui_url }}"
      }

You can create this with:

prefect automation create -f notify-completion.yaml

Best practices for returning data to an external tool - Small/JSON-sized results: return directly from the flow; your caller can retrieve with

flow_run.state.result()

as shown above. - Rich/large outputs: write an artifact (table/markdown/link) inside your flow and have your caller read it by following the UI link or a link artifact you create. - Example inside your flow:

from prefect import flow
    from prefect.artifacts import create_markdown_artifact, create_link_artifact

    @flow
    def my_flow():
        # ... compute ...
        create_markdown_artifact(key="summary", markdown="# Done!", description="Run summary")
        create_link_artifact(key="result", link="<https://my-bucket/result.json>", link_text="Download results")
        return {"ok": True}

Optional: decouple create vs wait If you want to create the run first and wait later (polling), you can use `wait_for_flow_run`: ``` from prefect.flow_runs import wait_for_flow_run # later, when you want to block until final state:

flow_run = wait_for_flow_run(flow_run_id, timeout=600) result = flow_run.state.result() ``` Docs and references - Run deployments: Run deployments - States and results: States and results - Artifacts: Artifacts - Automations: Automations - Webhook Block (for call-webhook actions): Webhook block Question for you - Do you want the LLM tool to block until the flow completes and return the result inline, or return immediately and receive a webhook callback later? I can share a minimal wrapper tailored to your tool’s interface once I know which pattern you prefer.

Here are the Prefect 3.x features intentionally built to get results back to the place that triggered a deployment, with concise examples. These cover both synchronous “wait and return” and asynchronous “call me back” patterns. 1) run_deployment with blocking and result retrieval - What it does: Triggers a deployment and (optionally) waits for completion. Returns a FlowRun whose final state contains the flow’s return value. - Use when: Your caller (e.g., an LLM tool) can wait for the run to finish and needs the return value inline.

from prefect.deployments import run_deployment

# Blocks until the flow finishes (or timeout), then returns the flow's return value
flow_run = run_deployment(
    name="my-flow/my-deployment",
    parameters={"x": 42},
    timeout=300,        # seconds; None = wait indefinitely, 0 = return immediately
    poll_interval=2
)

result = flow_run.state.result()  # raises if the run failed
print("Result:", result)

- Docs: Run deployments, States & results 2) wait_for_flow_run (decouple trigger and wait) - What it does: Lets you start a run now and “wait for it” later by ID. Good for services that create a run, return a correlation ID, then later wait/collect the result. - Use when: You want to return fast, then poll/wait later in your service.

from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run

# Kick off quickly
flow_run = run_deployment(name="my-flow/my-deployment", parameters={"x": 42}, timeout=0)

# Later: wait for it to finish and get the return value
flow_run = wait_for_flow_run(flow_run.id, timeout=600)
result = flow_run.state.result()

- Docs: Run deployments, States & results 3) Automations: call-webhook action (async callback to your system) - What it does: Calls your webhook when a flow run reaches a target state (Completed/Failed/etc.). Payload can include flow_run IDs, status, and UI links. - Use when: Your caller cannot block; you want Prefect to push results/metadata back to your system. Automation YAML example:

name: Notify on Flow Completion
enabled: true
trigger:
  type: event
  posture: Reactive
  expect:
    - prefect.flow-run.Completed
  threshold: 1
actions:
  - type: call-webhook
    block_document_id: "YOUR_WEBHOOK_BLOCK_ID"
    payload: |
      {
        "event": "flow_completed",
        "flow_run_id": "{{ flow_run.id }}",
        "flow_run_name": "{{ flow_run.name }}",
        "status": "{{ flow_run.state.name }}",
        "ui_url": "{{ flow_run|ui_url }}"
      }

Create it:

prefect automation create -f notify-completion.yaml

- Docs: Automations, Webhook block 4) Artifacts (structured results your caller can read) - What it does: Lets your flow publish structured outputs (markdown/table/link/progress) that are visible in the UI and retrievable by ID. - Use when: You want richer results than a plain return value (e.g., a summary, a link to a large file, or a table). Create artifacts in your flow:

from prefect import flow
from prefect.artifacts import create_markdown_artifact, create_link_artifact

@flow
def my_flow():
    # ... compute ...
    create_markdown_artifact(key="summary", markdown="# Processing complete")
    create_link_artifact(
        key="download",
        link="<https://my-bucket/results.json>",
        link_text="Download results"
    )
    return {"ok": True}

- Docs: Artifacts 5) Result persistence + state.result()/aresult()

- What it does: Prefect persists your flow’s return value and lets you load it from the final state via

state.result()

(sync) or

state.aresult()

(async). - Use when: You want reliable retrieval of return values even from remote workers.

# After you have a flow_run (from run_deployment or client lookup)
result = flow_run.state.result()     # sync
# or
# result = await flow_run.state.aresult()  # async

- Docs: States & results 6) Event subscribers (real-time streaming in your own process) - What it does: Subscribe to flow-run events programmatically and react to “Completed/Failed” in real-time. - Use when: You’re running a service that wants to stream updates for a specific run.

import asyncio
from prefect.events.subscribers import FlowRunSubscriber

async def watch(flow_run_id):
    async with FlowRunSubscriber(flow_run_id=flow_run_id) as sub:
        async for event in sub:
            if event.event == "prefect.flow-run.Completed":
                # call your callback or fetch results here
                break

- Docs: Automations (events model overview) Which one should you use? - Synchronous (run_deployment + state.result): Your LLM/tool can wait and wants the return value in the same request. - Asynchronous (Automations call-webhook): Your LLM/tool should return immediately and receive a callback later. - Hybrid (start now, wait later): Create then

wait_for_flow_run

later, or poll, or subscribe to events. If you tell me whether your LLM tool should block or receive a webhook callback, I’ll share a minimal wrapper tailored to that pattern.

Recommendation: use run_deployment and read the final state’s result Since your Python code will wait until the run finishes and then return success or error, the simplest and most intentional Prefect 3.x feature for this is: - Trigger with run_deployment - Wait (blocking) for completion - Read the result from the final state with state.result() Minimal example

from prefect.deployments import run_deployment

def run_and_get_result(deployment_name: str, parameters: dict):
    # Wait indefinitely for the flow to finish (set timeout to a number if you want a cap)
    flow_run = run_deployment(
        name=deployment_name,          # "flow-name/deployment-name"
        parameters=parameters,
        timeout=None,                  # None = wait until final; use e.g. 600 to cap at 10 min
        poll_interval=2
    )

    # Returns the flow's return value if Completed
    # Raises the original exception if the run Failed/Crashed/Cancelled
    result = flow_run.state.result()
    return result

Robust version (returns structured success/error)

from prefect.deployments import run_deployment

def run_and_report(deployment_name: str, parameters: dict, timeout_seconds: float | None = None):
    flow_run = run_deployment(
        name=deployment_name,
        parameters=parameters,
        timeout=timeout_seconds,  # None waits indefinitely
        poll_interval=2
    )

    try:
        value = flow_run.state.result()  # raises on failure by default
        return {
            "ok": True,
            "value": value,
            "flow_run_id": str(flow_run.id),
            "state": flow_run.state.type.value,
        }
    except Exception as exc:
        # If you want the exception object itself, call result(raise_on_failure=False)
        return {
            "ok": False,
            "error": str(exc),
            "flow_run_id": str(flow_run.id),
            "state": flow_run.state.type.value,
        }

Notes and best practices - Blocking behavior: timeout=None waits until a final state. If you set a timeout, you may get a non-final state back; in that case keep waiting or return a “still running” response. - Return value shape: Prefer JSON-friendly return values if an LLM will consume them. For large/rich outputs, write a link/table/markdown artifact inside your flow and return a compact summary. - Failure behavior: state.result() raises the original exception on failure. If you want to capture it without raising, use state.result(raise_on_failure=False). Docs - Run deployments: Run deployments - States and results: States & results