Configuring WebSocket for Real-Time Workflow Monitoring in ChatDev: A Complete Guide

ChatDev streams live workflow data to clients through a dedicated WebSocket layer consisting of WebSocketManager, WebSocketLogger, and WebSocketGraphExecutor, enabling real-time logs, artifact previews, and human-in-the-loop interactions.

The OpenBMB/ChatDev platform executes agent workflows as server-side graphs that require immediate visibility for users. By implementing a custom WebSocket architecture, the system broadcasts execution status, generated files, and interactive prompts without polling overhead. This guide explains the exact configuration and code patterns used in the repository to enable real-time workflow monitoring.

Architecture Overview

ChatDev treats every workflow as a graph of nodes executed by specialized services. To expose this execution stream in real time, the platform introduces a parallel WebSocket layer that runs alongside the core workflow engine. The architecture separates concerns into connection management, log streaming, and execution hooks, allowing any client to subscribe to live updates via a single persistent connection.

The system relies on five tightly-coupled components that handle distinct responsibilities:

• WebSocketManager – Central registry for active connections and message routing
• WebSocketLogger – Log adapter that pushes entries to connected clients
• WebSocketGraphExecutor – Extended executor that emits artifacts and prompts
• FastAPI /ws endpoint – HTTP upgrade handler that initializes sessions
• Global state singleton – Shared access point for the manager across modules

Core WebSocket Components

WebSocketManager

Located in [server/services/websocket_manager.py](https://github.com/OpenBMB/ChatDev/blob/main/server/services/websocket_manager.py), the WebSocketManager class maintains a dictionary of active_connections mapping session IDs to WebSocket objects. It provides three critical functions:

1. Connection handling – The connect(websocket) method accepts a FastAPI WebSocket instance, generates a UUID if none is provided, stores the connection, and immediately transmits a connection message containing the session ID.
2. Message dispatch – send_message_sync(session_id, payload) and broadcast methods deliver JSON envelopes to specific or all connected clients.
3. Lifecycle management – The manager handles heartbeat ping/pong sequences and graceful disconnection via disconnect(session_id), which cancels running workflows and cleans up resources.

WebSocketLogger

The [server/services/websocket_logger.py](https://github.com/OpenBMB/ChatDev/blob/main/server/services/websocket_logger.py) module extends the base WorkflowLogger to bridge internal logging with the WebSocket layer. When a log entry is generated by any node in the workflow graph, the logger invokes:

self.websocket_manager.send_message_sync(
    self.session_id, 
    {"type": "log", "data": log_entry.to_dict()}
)

This pushes structured log data—including timestamps, severity levels ([INFO], [DEBUG], [ERROR]), and messages—directly to the client console without buffering.

WebSocketGraphExecutor and WebPromptChannel

The [server/services/websocket_executor.py](https://github.com/OpenBMB/ChatDev/blob/main/server/services/websocket_executor.py) file defines WebSocketGraphExecutor, a subclass of GraphExecutor that injects two critical hooks:

• WorkspaceArtifactHook – Captures generated files, images, and artifacts via an emit_callback that calls self.artifact_dispatcher.emit_workspace_artifacts(artifacts). The underlying ArtifactDispatcher routes these to WebSocketManager.send_message_sync with type "artifact".
• WebPromptChannel – Intercepts nodes requiring human input. Instead of blocking on standard input, it sends a "prompt" message to the client and awaits a response through the WebSocket connection.

This design allows workflows to pause for human feedback while maintaining the live connection.

FastAPI Route and Global State

The WebSocket endpoint is exposed through [server/routes/websocket.py](https://github.com/OpenBMB/ChatDev/blob/main/server/routes/websocket.py):

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    manager = get_websocket_manager()
    session_id = await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.handle_message(session_id, data)
    except WebSocketDisconnect:
        manager.disconnect(session_id)

The get_websocket_manager() function, defined in [server/state.py](https://github.com/OpenBMB/ChatDev/blob/main/server/state.py), ensures a singleton pattern. Calling init_state() during server startup (typically in server_main.py) initializes this global registry before the FastAPI application begins accepting connections.

Message Flow and Communication Protocol

All WebSocket communication follows a uniform JSON envelope structure:

{
  "type": "<event>",
  "data": { ... }
}

Supported event types include connection, log, artifact, prompt, status, error, ping, and human_input.

The execution flow proceeds through these stages:

1. Connection establishment – Client opens ws://<host>/ws, receives session ID
2. Workflow initiation – Server creates WebSocketGraphExecutor for the session
3. Streaming execution – Logs emit via WebSocketLogger, artifacts via WorkspaceArtifactHook
4. Human interaction – WebPromptChannel sends prompt events; client responds with human_input payloads
5. Heartbeat – Client sends ping every 15 seconds; manager replies with pong
6. Termination – On disconnect, manager cleans up session and cancels active workflows

Implementation Guide

Server-Side Setup

Initialize the global state before starting the FastAPI application:

from server.state import init_state
from fastapi import FastAPI
from server.routes import websocket

def main():
    # Initialize singletons including WebSocketManager

    init_state()
    
    app = FastAPI()
    app.include_router(websocket.router)
    # Include other routers...

Ensure the WebSocketLogger is configured as the default logger for workflow runs, and that WorkflowRunService instantiates WebSocketGraphExecutor instead of the standard executor when a session ID is present.

Client-Side Integration

Connect to the endpoint and handle message types appropriately:

import { useEffect, useRef, useState } from "react";

export function useChatDevSocket(onLog, onArtifact, onPrompt) {
  const ws = useRef(null);
  const [sessionId, setSessionId] = useState(null);

  useEffect(() => {
    ws.current = new WebSocket(`ws://${window.location.host}/ws`);

    ws.current.onmessage = (event) => {
      const msg = JSON.parse(event.data);
      switch (msg.type) {
        case "connection":
          setSessionId(msg.data.session_id);
          break;
        case "log":
          onLog && onLog(msg.data);
          break;
        case "artifact":
          onArtifact && onArtifact(msg.data);
          break;
        case "prompt":
          onPrompt && onPrompt(msg.data);
          break;
      }
    };

    // Heartbeat to keep connection alive
    const heartbeat = setInterval(() => {
      ws.current?.send(JSON.stringify({type: "ping"}));
    }, 15000);

    return () => {
      clearInterval(heartbeat);
      ws.current?.close();
    };
  }, []);

  const send = (type, data) => {
    ws.current?.send(JSON.stringify({type, data}));
  };

  return { sessionId, send };
}

To respond to prompts from the workflow:

function handleUserAnswer(answer) {
  send("human_input", { answer });
}

Display logs in real time:

function LogConsole() {
  const [logs, setLogs] = useState([]);
  useChatDevSocket((log) => setLogs((prev) => [...prev, log]), null, null);

  return (
    <pre>
      {logs.map((l) => `${l.timestamp} [${l.level}] ${l.message}`).join("\n")}
    </pre>
  );
}

Summary

• WebSocketManager in server/services/websocket_manager.py acts as the central hub for connection registry, message broadcasting, and session lifecycle management.
• WebSocketLogger streams structured log entries to clients immediately as workflow nodes execute.
• WebSocketGraphExecutor extends the standard executor to emit artifacts and forward human-input prompts through the WebSocket connection.
• The /ws endpoint in server/routes/websocket.py handles HTTP upgrades and delegates to the manager singleton retrieved from server/state.py.
• All messages use a consistent JSON envelope with type and data fields, supporting events for logs, artifacts, prompts, and heartbeats.

Frequently Asked Questions

How does ChatDev handle WebSocket disconnections during active workflows?

When a client disconnects or the disconnect(session_id) method is called in WebSocketManager, the manager removes the connection from active_connections and cancels any running workflow associated with that session ID. This prevents orphaned processes and ensures resources are released immediately upon connection loss.

Can multiple clients monitor the same workflow execution simultaneously?

While the current architecture maps one session ID to one WebSocket connection per client, the WebSocketManager supports broadcasting via methods like send_message_sync or broadcast variants. To enable multi-client monitoring, you would modify WebSocketGraphExecutor to reference multiple session IDs or implement a pub/sub pattern in the manager that routes workflow events to all subscribed connections.

What is the purpose of the WebPromptChannel in the WebSocket architecture?

WebPromptChannel replaces standard input mechanisms for workflows requiring human interaction. When a node needs user feedback, it sends a "prompt" message through the WebSocket instead of blocking on console input. The client displays the prompt and returns the answer via a "human_input" message, which the WebSocketManager routes back into the paused graph execution, enabling true human-in-the-loop capabilities over the network.

How are generated files and artifacts transmitted to the client?

The WebSocketGraphExecutor installs a WorkspaceArtifactHook that captures any files, images, or documents created during node execution. The hook's emit_callback forwards these artifacts to ArtifactDispatcher.emit_workspace_artifacts, which ultimately calls WebSocketManager.send_message_sync with the event type set to "artifact". Clients receive the file metadata and binary data (or URLs) through the same persistent connection used for logs.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}