Connections in omnigent

omnigent is an open-source meta-harness: one orchestration layer that runs Claude Code, Codex, Cursor, and other coding agents under a common contract. While reading its source I kept getting lost in who connects to whom, so I put together a five-file runnable mock of just the connection fabric — a control-plane server, a runner that dials out instead of listening, and one harness subprocess per conversation. This post walks those connections end to end. Everything below uses a single conversation id, demo.

Terms first — who is the “client” here?

The words client and server mean different things at two different layers, and the roles invert between them. At the connection layer (who establishes the socket): the runner is the WebSocket client — it dials out — and the server is the WebSocket server — it accepts. At the application layer (who sends requests once the socket exists): the server is the HTTP client — it sends requests down the tunnel — and the runner is the HTTP server — it answers them, without listening on any port. This inversion is the whole point of the dial-out pattern: the machine that is easy to reach accepts the connection, then the machine that has something to ask uses it.

From here on, “client” and “server” are only used in the senses above — never as loose names for the machines.

Start with the clients

Let’s start with client.py. It starts 2 clients, which call the endpoint {SERVER}/sessions/demo/stream. Each watcher reads the SSE stream and sets an asyncio.Event when it sees event["type"] == "response.completed". Pretty standard POST/SSE flow.

async def watch(name: str, done: asyncio.Event) -> None:
    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("GET", f"{SERVER}/sessions/demo/stream") as resp:
            async for line in resp.aiter_lines():
                ...
                if event["type"] == "response.completed":
                    done.set()
                    return

One ordering detail: the watchers subscribe before anything is posted. The fan-out only reaches queues that are already registered, and the mock has no replay — a late subscriber simply misses earlier events.

Then, another client (it could as well be one of A or B) triggers /sessions/demo/messages with the following data — {"text": "hello omnigent"}.

The server

Up next is the omnigent server listening to the clients — a pretty standard uvicorn app with 2 endpoints, but inside the logic of these endpoints is magic.

The dial-out tunnel

Before discussing that: the runner may sit behind NAT or a firewall with no reachable address, so nothing can connect to it. The standard answer is the dialed-out WebSocket pattern — at startup, the runner opens one WebSocket to the server, and from then on the server sends its HTTP requests backward down that same socket. In the terms above: the runner is the connection-layer client; the server is the application-layer client. It is important to get this right, so take a moment to understand this pattern. The plumbing that makes it possible is runner_tunnel and _runner_client, defined in server.py.

runner_tunnel — accepts the runner’s dial-out, then starts a while loop which keeps reading from the websocket and puts the frames onto their corresponding queue. This is an important pattern: the same WS carries many concurrent, overlapping request/response exchanges, distinguished by the req_id.

while True:
    frame = common.decode(await websocket.receive_text())
    queue = pending.get(frame["req_id"])
    if queue is not None:
        await queue.put(frame)

_runner_client — defines an httpx.AsyncClient with a custom transport layer (TunnelTransport). Inside the transport layer, we read the request body, create a new req_id and an asyncio.Queue, and send the frame to the runner over the WS (send_text, to be precise). You will notice that this send is inside a lock. Why? At a given time, this uvicorn server can be sending multiple requests to the runner over the WS — two different awaits at the same time can corrupt the data stream and kill demuxing for every in-flight request on the shared tunnel. Once the message is sent, it awaits on the queue for the first incoming message (the RESPONSE_HEAD frame), then returns an httpx.Response with stream=_TunnelStream(req_id).

req_id = uuid.uuid4().hex
pending[req_id] = asyncio.Queue()
async with send_lock:
    await runner_ws.send_text(common.encode(frame))
head = await pending[req_id].get()   # RESPONSE_HEAD — guaranteed to arrive first
return httpx.Response(
    status_code=head["status"], stream=_TunnelStream(req_id), request=request
)

Now, back to the 2 endpoints.

/sessions/{sid}/messages

It creates an asyncio task (dispatch) which calls the runner using _runner_client. Note this task is not awaited here, so {"status": "dispatched"} is returned immediately. Inside dispatch, we use the client to call the runner’s /sessions/{sid}/turn. Internally, it uses the TunnelTransport and blocks on pending[req_id].get(). It takes exactly one item off the queue — the head, which is guaranteed to arrive first because the runner’s send() emits RESPONSE_HEAD before any body frame, and both the WebSocket and the demux loop preserve order. Then it hands the still-unread remainder of the queue to _TunnelStream(req_id) as the response body. When dispatch’s await client.post(...) reads the response (non-streaming httpx calls read the body automatically), that iterator drains the RESPONSE_BODY chunk — there’s your {"status":"done"} — and stops at RESPONSE_END, whose finally removes pending[req_id].

class _TunnelStream(httpx.AsyncByteStream):
    async def __aiter__(self):
        while True:
            frame = await pending[self.req_id].get()
            if frame["kind"] == common.RESPONSE_END:
                return
            yield frame["chunk"].encode()

/sessions/{sid}/stream

It creates a queue and appends it to the list of subscribers for this conversation. For the very first call for this conversation id, it triggers _relay_from_runner, which internally calls /sessions/{sid}/stream (runner.py) over the same transport layer as defined previously. Once a message is received from the runner’s stream, it is fanned out by appending it to each of the queues associated with this conversation id. Inside the route, gen() reads from the queue and returns an event stream.

One important point to highlight is that once _relay_from_runner is started, it continues to read data from the stream even when there are no subscribers left for this conversation id.

The runner

Now, this is what will run on our sandboxed VMs or local laptops, and it dials out to the omnigent server. For each WS message, it calls _dispatch, which internally creates a per-request ASGI scope plus send/receive callables, and then invokes the existing app. Note that this app is not hosted on uvicorn. Each frame is handled in its own asyncio.create_task, so a long-lived /stream request and short /turn requests overlap freely on the one socket — which is exactly why the shared send lock exists.

async with websockets.connect(SERVER_TUNNEL_URL) as ws:
    send_lock = asyncio.Lock()
    async for raw in ws:                 # parks here until a frame arrives
        asyncio.create_task(_dispatch(ws, common.decode(raw), send_lock))

receive and send

receive — it’s a callback the app calls to pull the request body; every subsequent call awaits forever on asyncio.get_running_loop().create_future(). This is important — otherwise Starlette cancels the /stream route the moment its one request body has been read, before any SSE body is sent.

async def receive() -> dict:
    nonlocal body_sent
    if not body_sent:
        body_sent = True
        return {"type": "http.request", "body": body, "more_body": False}
    return await asyncio.get_running_loop().create_future()   # parks forever

send — it sends the route’s response back to the omnigent server as WS frames, using the same asyncio lock pattern to prevent data corruption.

async def send(message: dict) -> None:
    if message["type"] == "http.response.start":
        async with send_lock:
            await ws.send(common.encode(
                common.response_head_frame(req_id, message["status"])))
    elif message["type"] == "http.response.body":
        chunk = message.get("body", b"")
        if chunk:
            async with send_lock:
                await ws.send(common.encode(
                    common.response_body_frame(req_id, chunk.decode())))
        if not message.get("more_body", False):
            async with send_lock:
                await ws.send(common.encode(
                    common.response_end_frame(req_id)))

The primary reason for defining the send/receive methods ourselves is that httpx.ASGITransport collects every message in its buffer and returns an httpx.Response only when the app function returns. So, if you look at the /stream endpoint, it never returns due to the while loop — the messages sit in the buffer but are never returned. A permanent hang. For a more detailed answer, check STREAMING.md.

/sessions/{sid}/turn

It reads the request and creates a harness instance if there is none for the conversation id. The harness process is also a uvicorn FastAPI server, but the runner and harness connect via a Unix socket. It provides multiple benefits: user isolation, as no other user can read the /tmp/omnigent-mock-<uid> directory (created with 0700); no open ports, so no firewall concerns; and no race conditions on port numbers as multiple harness instances are spawned — each conversation derives its own socket path instead of hunting for a free port.

harness_procs[sid] = subprocess.Popen(
    [sys.executable, str(HARNESS_PATH), "--uds", str(sock_path)]
)
await _wait_for_socket(sock_path)
client = httpx.AsyncClient(
    transport=httpx.AsyncHTTPTransport(uds=str(sock_path)),
    base_url="http://harness", timeout=None,
)

Up next, we call the harness server’s /turn endpoint with the user’s request, read data from the stream, and append it to the queue (which the /stream route consumes). One nuance here: only once the entire turn finishes does this route return {"status": "done"} — and only then is the http.response.start sent from the runner to the server (refer to the table above).

Also, in this route, when reading the response from the harness, if the event is a tool call, the runner runs that tool and sends back the result via a POST to /events. The identifier in this case is the tool call id.

/sessions/{sid}/stream

It gets the queue for the conversation id, reads data from it, and sends it back as a StreamingResponse. Each message from the StreamingResponse goes through the custom send function, which forwards it to the omnigent server as WS frames.

The harness

It is a uvicorn server, as mentioned earlier, with 2 endpoints.

/turn — it receives the user’s message and calls the agent (_vendor_agent). The agent’s response is streamed back to the runner over SSE. Mid-stream, the agent asks for a tool and parks on a future:

yield _sse({"type": "response.function_call", "call_id": call_id, "name": "get_time"})

future = asyncio.get_running_loop().create_future()
pending_tool_calls[call_id] = future
result = await future            # resolved later by POST /events

/events — it receives the tool result from the runner. Importantly, the agent awaits the result for this tool call via that future — the result arrives as a posted event, not as a return value.

future = pending_tool_calls.pop(body["call_id"], None)
if future is not None and not future.done():
    future.set_result(body["result"])

A dedicated harness is created for each conversation — and reused across its turns.

References

• The runnable mock: github.com/lakhand7/omnigent-mock — five files, one command (python demo.py) runs the whole narrative end to end. Its STREAMING.md covers the buffering/deadlock story in depth.
• The real thing: omnigent — the patterns here are miniatures of its runner/transports/ws_tunnel/, runtime/harnesses/, and server session routes.