Streaming the Vibe: Real-time Agentic UX with FastAPI WebSockets

Master the art of low-latency Agentic UX using Python, Pydantic AI, and LangGraph to build fluid, high-intent AI interfaces.

In the era of Vibe Coding, the distance between thought and execution has collapsed. By 2026, the traditional "request-response" cycle of the web feels like a relic of a slower age. Users no longer want to wait for a spinning loader while an LLM processes a complex query; they want to see the agent thinking, pivoting, and constructing the UI in real-time.

Agentic UX is the manifestation of this shift. It is the transition from static interfaces to living, breathing software that streams its reasoning and intermediate states directly to the user. To achieve this, we move beyond REST and embrace the bi-directional power of FastAPI WebSockets, orchestrated by the precision of Pydantic AI and the stateful logic of LangGraph.

1. The Architectural Blueprint: FastAPI + WebSockets

To stream a "vibe," your backend must stay connected. Standard HTTP is too rigid for the asynchronous nature of multi-agent workflows. FastAPI’s implementation of WebSockets allows us to push updates to the client the moment a tool is called or a thought is generated.

The Connection Manager

We need a robust way to handle the state of our "Vibe Stream."

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

class VibeConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def stream_update(self, message: dict, websocket: WebSocket):
        await websocket.send_json(message)

manager = VibeConnectionManager()
app = FastAPI()

@app.websocket("/v1/stream/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # Agentic orchestration starts here
    except WebSocketDisconnect:
        manager.disconnect(websocket)

2. The Brain: Structured Reasoning with Pydantic AI and Gemini

Vibe Coding isn't about raw text; it’s about intent. We use Pydantic AI to wrap the Gemini 1.5 Pro API, ensuring that every thought the agent has conforms to a schema that our frontend can actually render.

Gemini’s massive context window allows our agent to remember the "vibe" of the entire conversation, while Pydantic AI enforces type safety on the streaming output.

from pydantic_ai import Agent
from pydantic import BaseModel

class AgentThought(BaseModel):
    action: str
    thought_process: str
    ui_component: str  # e.g., "Chart", "Table", "Markdown"
    data: dict

vibe_agent = Agent(
    'google-gla:gemini-1.5-pro',
    result_type=AgentThought,
    system_prompt="You are a Vibe Orchestrator. Stream your reasoning and UI suggestions."
)

3. The Orchestrator: LangGraph for Stateful Flows

If Pydantic AI is the brain, LangGraph is the nervous system. Agentic UX requires the ability to loop, backtrack, and branch. LangGraph allows us to define a graph where each node represents a step in the agent's workflow—searching, analyzing, or rendering—and stream the state of that graph back through our WebSocket.

Implementing a Stateful Stream

from langgraph.graph import StateGraph, END

def agent_node(state):
    # Logic to invoke Pydantic AI and Gemini
    return {"messages": [vibe_agent.run_sync(state["input"])]}

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.set_entry_point("agent")
workflow.add_edge("agent", END)

# Compile the graph
app_vibe = workflow.compile()

4. Streaming the UX: Real-time UI Evolution

The magic happens when we bridge the gap. As LangGraph moves through nodes, we use the manager.stream_update method to send JSON packets to the frontend.

Instead of a terminal log, the user sees a Dynamic UI:

1. 0ms: User sends intent.
2. 200ms: Agent sends "Thought" packet (UI renders a skeleton).
3. 800ms: Agent calls a tool (UI renders a progress bar).
4. 1500ms: Agent receives data (UI renders a high-quality Recharts component).

The Streaming Logic

async def run_agentic_flow(user_input: str, websocket: WebSocket):
    async for event in app_vibe.astream({"input": user_input}):
        for node, state in event.items():
            # Extract the 'vibe' and push to client
            payload = {
                "node": node,
                "content": state["messages"][-1].content,
                "timestamp": "now"
            }
            await manager.stream_update(payload, websocket)

The Paradigm Shift: Why "Vibe Coding" Wins

Traditional development focuses on Imperative UX—you build every button and route. Agentic UX is Declarative. You define the capabilities, and the agent constructs the path in real-time.

Practical "Vibe Check": Implementing Today

To get this stack running immediately, follow this checklist:

1. FastAPI Skeleton: Set up your WebSocket routes and connection manager.
2. Schema Definition: Use Pydantic to define exactly what your "UI packets" look like. Never stream raw, unvalidated strings.
3. Graph Logic: Map out your agent's decision tree in LangGraph. Start with a simple "Think -> Tool -> Respond" loop.
4. Gemini Integration: Use Gemini 1.5 Flash for speed-sensitive UX and Pro for deep reasoning tasks.
5. Frontend Listener: Ensure your React/Vue/Svelte frontend is ready to parse incoming JSON and map it to a library of pre-built components (shadcn/ui is the gold standard here).

Conclusion: Engineering the Future of Intent

Streaming the vibe is more than a technical hurdle; it’s a commitment to a superior user experience. By combining the speed of FastAPI, the structure of Pydantic AI, the reasoning of Gemini, and the orchestration of LangGraph, you aren't just building an app—you're building a collaborator.

In the world of Agentic UX, the UI is no longer a destination; it is a real-time conversation between human intent and machine execution.

Elevate Your Workflow

Is your organization ready to transition from legacy workflows to autonomous agentic systems? Azura AI specializes in custom Document Workflow Automation and high-performance AI agent orchestration.

Optimize your Vibe with Azura AI today.