https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> can you give me an example how to run...
# ask-marvin
r
@Marvin can you give me an example how to run a prefectAgent in a fastapi endpoint with SSE?
m
✅ thought for 137.9 seconds
Short answer: there isn’t a 
prefectAgent
in Prefect 3.x. If you mean: - Prefect 2.x agent → that concept was replaced by workers in Prefect 3.x - some kind of AI agent → Prefect doesn’t provide a built-in 
prefectAgent
API for that So for FastAPI + SSE, the usual Prefect pattern is: 1. start a flow run or task run from the endpoint 2. stream status/log/progress updates back over SSE by polling Prefect or subscribing to updates A simple working pattern looks like this. Example: start a Prefect flow and stream status via SSE
Copy code
python
import asyncio
import json
from uuid import UUID

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from prefect import flow
from prefect.client.orchestration import get_client

app = FastAPI()


@flow
async def example_flow(name: str):
    for i in range(5):
        print(f"step {i} for {name}")
        await asyncio.sleep(1)
    return {"message": f"done for {name}"}


@app.post("/runs")
async def create_run(name: str):
    # Creates a flow run in Prefect and returns its id
    state = await example_flow.with_options(return_state=True)(name)

    flow_run_id = state.state_details.flow_run_id
    return {"flow_run_id": str(flow_run_id)}


@app.get("/runs/{flow_run_id}/events")
async def stream_run(flow_run_id: UUID):
    async def event_generator():
        async with get_client() as client:
            last_state = None

            while True:
                flow_run = await client.read_flow_run(flow_run_id)
                state = flow_run.state

                payload = {
                    "flow_run_id": str(flow_run_id),
                    "state_type": str(state.type) if state else None,
                    "state_name": state.name if state else None,
                }

                if payload != last_state:
                    yield f"data: {json.dumps(payload)}\n\n"
                    last_state = payload

                if state and state.is_final():
                    break

                await asyncio.sleep(1)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )
A browser client could consume it like this:
Copy code
javascript
const evtSource = new EventSource("/runs/<flow_run_id>/events");

evtSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data);
};
A more realistic pattern The example above runs the flow inline from the API request. That’s okay for demos, but for production it’s usually better to: - deploy the flow - trigger a deployment run from FastAPI - stream the resulting flow run state via SSE That way your API doesn’t execute the flow process directly. The flow trigger endpoint would look more like:
Copy code
python
from fastapi import FastAPI
from prefect.client.orchestration import get_client

app = FastAPI()

@app.post("/deployments/{deployment_id}/runs")
async def create_deployment_run(deployment_id: str):
    async with get_client() as client:
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters={"name": "marvin"},
        )
    return {"flow_run_id": str(flow_run.id)}
Then use the same SSE endpoint to poll 
read_flow_run
. Important caveats - Don’t try to run a Prefect worker inside a FastAPI endpoint. Workers are long-lived processes, not request-scoped objects. - If you were thinking of the old Prefect 2.x “agent”, in 3.x you should use a worker started separately. - SSE is just the transport. Prefect doesn’t have a special “SSE agent” API. - If you want live logs, you’ll need an additional mechanism beyond simple state polling. If you want, I can give you either: 1. a complete FastAPI example that triggers a deployment and streams status, or
2. a FastAPI example that also streams Prefect logs over SSE. Relevant docs: - Deploy via Python - Prefect workers - Prefect orchestration client
r
there is clearly a PrefectAgent provided by pydantic ai @Marvin
m
✅ thought for 134.6 seconds
You’re right — thanks for the correction. 
PrefectAgent
does exist, but it’s provided by 
pydantic-ai
, not by Prefect core itself. The import is:
Copy code
python
from pydantic_ai.durable_exec.prefect import PrefectAgent, TaskConfig
So the distinction is: - Prefect core does not define a 
prefectAgent
- PydanticAI provides 
PrefectAgent
as a Prefect-backed durable execution wrapper If your goal is FastAPI + SSE, you can use 
PrefectAgent
, but the best pattern depends on what you want to stream: - token / text streaming to the client → use 
agent.run_stream(...)
- durable retries / tool orchestration / observability in Prefect → wrap with 
PrefectAgent
Here’s a concrete FastAPI SSE example using 
PrefectAgent
. Example: FastAPI endpoint with SSE using 
PrefectAgent
Copy code
python
import json

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectAgent, TaskConfig

app = FastAPI()

base_agent = Agent(
    "openai:gpt-4o",
    name="demo-agent",
    system_prompt="You are a helpful assistant."
)

agent = PrefectAgent(
    base_agent,
    name="demo-agent",
    model_task_config=TaskConfig(
        retries=2,
        retry_delay_seconds=[1, 2],
        timeout_seconds=60,
    ),
    tool_task_config=TaskConfig(
        retries=2,
        timeout_seconds=30,
    ),
)

@app.get("/stream")
async def stream(prompt: str):
    async def event_generator():
        try:
            async with agent.run_stream(prompt) as response:
                async for chunk in response.stream():
                    yield f"data: {json.dumps({'type': 'chunk', 'content': chunk})}\n\n"

                output = response.get_output()
                yield f"data: {json.dumps({'type': 'done', 'output': output})}\n\n"

        except Exception as exc:
            yield f"data: {json.dumps({'type': 'error', 'error': str(exc)})}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )
Client-side example
Copy code
javascript
const es = new EventSource("/stream?prompt=Write%20a%20haiku%20about%20data");

es.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data);
};
If you want tools too
Copy code
python
import json

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic_ai import Agent, RunContext
from pydantic_ai.durable_exec.prefect import PrefectAgent, TaskConfig

app = FastAPI()

def get_weather(ctx: RunContext[None], city: str) -> str:
    return f"The weather in {city} is sunny."

base_agent = Agent(
    "openai:gpt-4o",
    name="weather-agent",
    system_prompt="Use tools when helpful.",
    tools=[get_weather],
)

agent = PrefectAgent(
    base_agent,
    name="weather-agent",
    model_task_config=TaskConfig(retries=2, timeout_seconds=60),
    tool_task_config=TaskConfig(retries=2, timeout_seconds=30),
)

@app.get("/weather-stream")
async def weather_stream(prompt: str):
    async def event_generator():
        try:
            async with agent.run_stream(prompt) as response:
                async for chunk in response.stream():
                    yield f"data: {json.dumps({'content': chunk})}\n\n"

                yield f"data: {json.dumps({'done': True, 'output': response.get_output()})}\n\n"
        except Exception as exc:
            yield f"data: {json.dumps({'error': str(exc)})}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")
Important caveat For pure SSE token streaming, you may not need 
PrefectAgent
at all. A plain PydanticAI 
Agent
is often simpler and more responsive:
Copy code
python
from pydantic_ai import Agent
Use 
PrefectAgent
when you specifically want: - Prefect-backed durability
- retries around model/tool execution - Prefect observability for those runs So the rule of thumb is: - interactive streaming endpoint → base 
Agent
is usually simplest - production durable AI workflow → 
PrefectAgent
There don’t seem to be Prefect GitHub discussions specifically covering 
PydanticAI PrefectAgent + FastAPI + SSE
, but the relevant pieces are here: - PydanticAI docs - Prefect docs If you want, I can give you a full copy-paste FastAPI app with: 1. 
sse-starlette
support, 2. typed request/response models, 3. a tool example, 4. and a version that emits structured SSE events like 
event: token
, 
event: done
, 
event: error
.
Open in Slack
PreviousNext
Powered by