https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> i want to steer a PrefectAgent (pydan...
# ask-marvin
r
@Marvin i want to steer a PrefectAgent (pydantic ai) that runs in a deployment. Is that possible/can you give me an example. My stack is fastapi, pydantic ai and prefect via ecs
m
✅ thought for 189.5 seconds
Yes — this is possible in Prefect 3.x, and your stack is a good fit for it. The clean pattern is: 1. FastAPI starts a Prefect deployment run 2. The deployed flow runs your PydanticAI agent on ECS 3. Your app steers the running flow by sending typed inputs to that flow run 4. The flow can either: - keep running and poll for steering input with 
receive_input(...)
, or - explicitly pause for human input with 
pause_flow_run(...)
/ 
suspend_flow_run(...)
For an interactive agent, I’d usually recommend 
receive_input
for steering and 
pause_flow_run
only for approval gates. Relevant Prefect primitives: - 
prefect.input.send_input(run_input, flow_run_id, ...)
- 
prefect.input.receive_input(input_type, ...)
- 
prefect.flow_runs.pause_flow_run(...)
- 
prefect.flow_runs.suspend_flow_run(...)
I also verified 
prefect deploy
exists in 3.x. Docs / references: - Prefect docs - Deployments - Flows - Work pools - Run input API - Flow run controls A practical architecture for your stack - FastAPI = control plane / user API - Prefect deployment = orchestrated agent run - ECS work pool = execution environment - PydanticAI agent = the reasoning loop inside the flow So instead of trying to “remote control a process directly”, you let Prefect orchestrate the process and use Prefect’s flow-run input channel as the steering interface. Example flow of control:
Copy code
text
Client/UI
  -> FastAPI POST /agents/start
  -> create Prefect flow run from deployment
  -> ECS task starts and runs agent flow

Client/UI
  -> FastAPI POST /agents/{flow_run_id}/steer
  -> send_input(...) to that running flow run

Running flow
  -> receive_input(...) reads commands
  -> updates agent behavior / prompt / next action
Minimal example First, define typed steering messages.
Copy code
python
from pydantic import BaseModel
from prefect.input import RunInput

class AgentCommand(RunInput):
    action: str   # "guide", "stop", "approve"
    message: str = ""

class AgentUpdate(RunInput):
    status: str
    detail: str
Now the Prefect flow that runs the agent: ```python import asyncio from prefect import flow, get_run_logger from prefect.context import get_run_context from prefect.input import receive_input, send_input from pydantic_ai import Agent class AgentCommand(RunInput): action: str message: str = "" class AgentUpdate(RunInput): status: str detail: str agent = Agent( "openai:gpt-4o-mini", system_prompt="You are a helpful agent. Incorporate operator guidance when provided." ) @flow async def steered_agent_flow(initial_prompt: str): logger = get_run_logger() flow_run_id = get_run_context().flow_run.id current_goal = initial_prompt stopped = False step = 0 while not stopped: step += 1 # Check for steering input without blocking forever command_handler = receive_input( AgentCommand, flow_run_id=flow_run_id, timeout=1, poll_interval=0.5, raise_timeout_error=False, ) command = next(command_handler, None) if command: if command.action == "stop": stopped = True await send_input( AgentUpdate(status="stopping", detail="Stop requested by operator"), flow_run_id=flow_run_id, ) break if command.action == "guide": current_goal = f"{current_goal}\n\nOperator guidance: {command.message}" await send_input( AgentUpdate(status="guided", detail=f"Applied guidance: {command.message}"),
flow_run_id=flow_run_id, ) await send_input( AgentUpdate(status="thinking", detail=f"Running step {step}"), flow_run_id=flow_run_id, ) result = await agent.run(current_goal) await send_input( AgentUpdate(status="result", detail=str(result.output)), flow_run_id=flow_run_id, ) # Example loop exit condition if step >= 5: break await asyncio.sleep(2) return {"final_goal": current_goal, "steps": step}
Copy code
Then in *FastAPI*, start and steer the deployment.

```python
from uuid import UUID
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from prefect.client.orchestration import get_client
from prefect.input import send_input

app = FastAPI()

class StartRequest(BaseModel):
    prompt: str

class SteerRequest(BaseModel):
    action: str
    message: str = ""

class AgentCommand(RunInput):
    action: str
    message: str = ""

@app.post("/agents/start")
async def start_agent(req: StartRequest):
    async with get_client() as client:
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id="YOUR_DEPLOYMENT_ID",
            parameters={"initial_prompt": req.prompt},
        )
    return {"flow_run_id": str(flow_run.id)}

@app.post("/agents/{flow_run_id}/steer")
async def steer_agent(flow_run_id: UUID, req: SteerRequest):
    async with get_client() as client:
        flow_run = await client.read_flow_run(flow_run_id)

    if not flow_run.state or not flow_run.state.is_running():
        raise HTTPException(status_code=400, detail="Flow run is not running")

    await send_input(
        AgentCommand(action=req.action, message=req.message),
        flow_run_id=flow_run_id,
        sender="fastapi",
    )

    return {"ok": True}
When to use 
pause_flow_run
instead If you want the agent to stop and wait for a human decision, use:
Copy code
python
from prefect.flow_runs import pause_flow_run

approval = pause_flow_run(wait_for_input=AgentCommand, timeout=3600)
That’s better for: - approval/reject - selecting one of N plans - editing a draft before proceeding But for “live steering” of an agent loop, 
receive_input
is usually the better fit. ECS-specific notes Running this in ECS is fine. The main things to think about are: - Long-running flow runs: if your agent loop stays alive for a while, make sure the ECS task sizing/timeouts fit that usage. - Networking: the task must be able to reach your Prefect API / Prefect Cloud. - Stateless workers: treat the flow as the authoritative state machine; don’t rely on local container memory beyond the current run. - Streaming UI updates: if you want a live UI, you’ll likely want your FastAPI app to poll Prefect state / artifacts / logs, or use a side channel for updates. Recommended production pattern for your stack I’d suggest this shape: 1. FastAPI endpoint starts deployment 2. Flow run ID is stored in your app/db 3. Agent flow loops: - execute one reasoning/action step - check 
receive_input(...)
for operator commands - persist meaningful intermediate state externally if needed 4. FastAPI sends steer commands with 
send_input(...)
5. Optional: - use 
pause_flow_run(...)
for explicit human approvals - use Prefect artifacts/logs for traceability One important design recommendation Don’t try to run a single infinitely interactive chatbot session as one giant monolithic flow unless you really need that. Often a better approach is: - one deployment run per “agent job” or “reasoning session” - steering allowed during that run - results/state persisted outside the container That plays nicely with ECS and Prefect’s orchestration model. Deployment In Prefect 3.x, use 
prefect deploy
or Python deployment APIs — not old 2.x methods. CLI:
Copy code
bash
prefect deploy --help
If you want, I can give you a full working example tailored to your stack, for either: 1. FastAPI + Prefect Cloud + ECS work pool + PydanticAI, or 2. a human-in-the-loop approval agent using 
pause_flow_run
. If you want, I can draft the exact files: - 
main.py
for FastAPI - 
flows/agent_flow.py
- 
prefect.yaml
- ECS work pool assumptions
r
@Marvin can you updae this example to work with SSE
m
✅ thought for 345.7 seconds
Yes — and for SSE I’d slightly change the architecture from the earlier example. Important distinction: - Use 
send_input
/ 
receive_input
 for steering commands into the running flow - Use Prefect events (or another outbound channel) for streaming updates out to the client Why: 
receive_input(...)
is polling-oriented and works well for control messages, but it’s not the best primitive for low-latency outbound streaming to browsers. So the recommended pattern is:
Copy code
text
Browser
  <- SSE from FastAPI
FastAPI
  <- Prefect events websocket
  -> Prefect send_input() for steering
Running Prefect flow on ECS
  -> emit_event() for progress / token-ish updates
  <- receive_input() for operator steering
If you want, I can also show a simpler polling-only SSE version, but this is the more “real-time” Prefect-native approach. Docs: - Prefect docs - Events - Python events API - Run input API --- Updated example with SSE ## 1. Prefect flow: steerable agent that emits events ```python import asyncio from typing import Optional from pydantic import BaseModel from pydantic_ai import Agent from prefect import flow, get_run_logger from prefect.context import get_run_context from prefect.events import emit_event from prefect.input import RunInput, receive_input class AgentCommand(RunInput): action: str # "guide", "stop" message: str = "" agent = Agent( "openai:gpt-4o-mini", system_prompt="You are a helpful agent. Incorporate operator guidance when provided." ) def emit_agent_event(event: str, resource_id: str, **payload): emit_event( event=event, resource={ "prefect.resource.id": resource_id, "prefect.resource.role": "flow-run", }, payload=payload, ) @flow(name="steered-agent-flow") async def steered_agent_flow(initial_prompt: str): logger = get_run_logger() flow_run = get_run_context().flow_run flow_run_id = str(flow_run.id) resource_id = f"prefect.flow-run.{flow_run_id}" current_goal = initial_prompt stopped = False step = 0 emit_agent_event( "agent.session.started", resource_id, flow_run_id=flow_run_id, message="Agent session started", initial_prompt=initial_prompt, ) while not stopped: step += 1 command_stream = receive_input( AgentCommand, flow_run_id=flow_run.id, timeout=1, poll_interval=0.5, raise_timeout_error=False, ) command = next(command_stream, None) if command: if command.action == "stop": emit_agent_event( "agent.command.received", resource_id, flow_run_id=flow_run_id, action="stop", message="Stop requested", ) stopped = True break if command.action == "guide": current_goal = f"{current_goal}\n\nOperator guidance: {command.message}" emit_agent_event( "agent.command.received", resource_id, flow_run_id=flow_run_id, action="guide", guidance=command.message, ) emit_agent_event( "agent.step.started", resource_id, flow_run_id=flow_run_id, step=step, goal=current_goal, ) result = await agent.run(current_goal) output_text = str(result.output) emit_agent_event( "agent.step.completed", resource_id, flow_run_id=flow_run_id, step=step, output=output_text, )
logger.info(f"Step {step}: {output_text}") if step >= 5: break await asyncio.sleep(2) emit_agent_event( "agent.session.finished", resource_id, flow_run_id=flow_run_id, steps=step, stopped=stopped, ) return {"steps": step, "stopped": stopped, "final_goal": current_goal} ``` --- ## 2. FastAPI app: start runs, steer runs, and expose SSE This example does three things: - starts the deployment - sends steering commands with 
send_input
- subscribes to Prefect events and relays them as SSE ```python import asyncio import json from uuid import UUID from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel import websockets from prefect.client.orchestration import get_client from prefect.input import RunInput, send_input PREFECT_API_URL = "https://api.prefect.cloud/api/accounts/ACCOUNT/workspaces/WORKSPACE" PREFECT_API_KEY = "YOUR_PREFECT_API_KEY" app = FastAPI() class StartRequest(BaseModel): prompt: str class SteerRequest(BaseModel): action: str message: str = "" class AgentCommand(RunInput): action: str message: str = "" @app.post("/agents/start") async def start_agent(req: StartRequest): async with get_client() as client: flow_run = await client.create_flow_run_from_deployment( deployment_id="YOUR_DEPLOYMENT_ID", parameters={"initial_prompt": req.prompt}, ) return {"flow_run_id": str(flow_run.id)} @app.post("/agents/{flow_run_id}/steer") async def steer_agent(flow_run_id: UUID, req: SteerRequest): async with get_client() as client: flow_run = await client.read_flow_run(flow_run_id) if not flow_run.state or not flow_run.state.is_running(): raise HTTPException(status_code=400, detail="Flow run is not running") await send_input( AgentCommand(action=req.action, message=req.message), flow_run_id=flow_run_id, sender="fastapi", ) return {"ok": True} def prefect_events_ws_url() -> str: if PREFECT_API_URL.startswith("https://"): return PREFECT_API_URL.replace("https://", "wss://") + "/events/out" if PREFECT_API_URL.startswith("http://"): return PREFECT_API_URL.replace("http://", "ws://") + "/events/out" raise ValueError("Unsupported PREFECT_API_URL") @app.get("/agents/{flow_run_id}/events") async def stream_agent_events(flow_run_id: UUID): async def event_generator(): ws_url = prefect_events_ws_url() resource_id = f"prefect.flow-run.{flow_run_id}" yield f"event: ready\ndata: {json.dumps({'flow_run_id': str(flow_run_id)})}\n\n" async with websockets.connect( ws_url, additional_headers={"Authorization": f"Bearer {PREFECT_API_KEY}"}, ) as ws: filter_message = { "type": "filter", "filter": { "event": { "name": [ "agent.session.started", "agent.command.received", "agent.step.started", "agent.step.completed", "agent.session.finished", "prefect.flow-run.Running", "prefect.flow-run.Completed", "prefect.flow-run.Failed", "prefect.flow-run.Crashed", ] }, "resource": { "id": [resource_id] } } } await ws.send(json.dumps(filter_message)) while True: try: message = await asyncio.wait_for(ws.recv(), timeout=20) data = json.loads(message) yield f"event: prefect\ndata: {json.dumps(data)}\n\n"
except asyncio.TimeoutError: yield "event: ping\ndata: {}\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", }, )
Copy code
---

## 3. Browser/client example

```html
<!DOCTYPE html>
<html>
  <body>
    <button onclick="start()">Start</button>
    <button onclick="guide()">Guide</button>
    <button onclick="stopAgent()">Stop</button>

    <pre id="log"></pre>

    <script>
      let flowRunId = null;
      let es = null;

      function log(msg) {
        document.getElementById("log").textContent += msg + "\n";
      }

      async function start() {
        const res = await fetch("/agents/start", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ prompt: "Research the best ECS deployment pattern for AI agents" })
        });

        const data = await res.json();
        flowRunId = data.flow_run_id;
        log("Started flow run: " + flowRunId);

        es = new EventSource(`/agents/${flowRunId}/events`);

        es.addEventListener("ready", (e) => {
          log("SSE ready: " + e.data);
        });

        es.addEventListener("prefect", (e) => {
          log("EVENT: " + e.data);
        });

        es.addEventListener("ping", () => {
          log("ping");
        });
      }

      async function guide() {
        if (!flowRunId) return;

        await fetch(`/agents/${flowRunId}/steer`, {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            action: "guide",
            message: "Focus on cost efficiency and low operational overhead"
          })
        });

        log("Sent guidance");
      }

      async function stopAgent() {
        if (!flowRunId) return;

        await fetch(`/agents/${flowRunId}/steer`, {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({
            action: "stop",
            message: ""
          })
        });

        log("Sent stop");
      }
    </script>
  </body>
</html>
--- Notes / caveats ## Why not stream directly from 
send_input
? Because 
send_input
/ 
receive_input
are better modeled as control messages, not high-frequency event streaming. They’re great for: - stop - revise plan - approve - inject operator guidance They’re less ideal for: - frequent token streaming - sub-second status fanout to browsers ## If you want token-level LLM streaming If by SSE you mean actual token streaming from PydanticAI, I’d recommend: - stream tokens from FastAPI directly if the LLM call originates there, or - have the flow emit coarser-grained progress events and keep true token streaming outside Prefect orchestration Running token-by-token through Prefect events is usually possible in theory but often too chatty for a production orchestration layer. ## ECS compatibility This works fine on ECS as long as: - the flow container can reach Prefect API / Cloud - your FastAPI service can reach Prefect’s events websocket - you size tasks for long-lived runs if the agent is interactive ## Security Do not expose the Prefect API key to the browser. Keep the websocket connection in FastAPI, and expose only your SSE endpoint to clients. --- What I’d recommend for your stack For FastAPI + PydanticAI + Prefect on ECS: - FastAPI owns user sessions and SSE - Prefect deployment owns orchestration and durable execution - RunInput handles steering into the flow - Prefect events handle outbound status streaming - optional: store agent transcript/checkpoints in Redis/Postgres/S3 if you want replay/history If you want, I can now turn this into a more complete working project layout with: - 
main.py
- 
flows/agent_flow.py
- 
prefect.yaml
- example ECS deployment assumptions - and an SSE endpoint using 
sse-starlette
instead of raw 
StreamingResponse
7 Views
Open in Slack
PreviousNext
Powered by