AgentOS Teams, Flows & WebSocket

Invoke teams and flows over HTTP, and chat with agents over a WebSocket

AgentOS routes every team and flow you pass in as its own HTTP endpoint, and exposes a chat WebSocket that streams token deltas from the agent.

Quick Start

1. Serve teams, flows, and WebSocket chat

   Pass teams= and flows= to AgentOS — each one becomes its own route, and /api/chat/stream is always available.

   from praisonai import AgentOS
   from praisonaiagents import Agent, Task, AgentTeam, AgentFlow
   
   team = AgentTeam(
       name="research_team",
       agents=[Agent(name="researcher", instructions="Research the topic")],
       tasks=[Task(description="Summarise the input", agent="researcher")],
   )
   
   flow = AgentFlow(
       name="writer_flow",
       steps=[Agent(name="writer", instructions="Write a short reply")],
   )
   
   AgentOS(
       agents=[Agent(name="assistant", instructions="Be helpful")],
       teams=[team],
       flows=[flow],
   ).serve(port=8000)

2. Call every surface

   Teams and flows are plain POST endpoints; the chat stream is a WebSocket.

   # Team
   curl -X POST http://localhost:8000/api/teams/research_team/run \
     -H "Content-Type: application/json" \
     -d '{"message": "quantum computing basics"}'
   
   # Flow
   curl -X POST http://localhost:8000/api/flows/writer_flow/run \
     -H "Content-Type: application/json" \
     -d '{"message": "Draft a one-line haiku"}'
   
   # WebSocket chat (streaming by default)
   websocat ws://localhost:8000/api/chat/stream
   > {"message": "Hello", "agent_name": "assistant"}
   < {"delta": "Hi"}
   < {"delta": " there!"}
   < {"done": true}

How It Works

A team or flow call acquires that instance’s lock, invokes the async entry point first, and offloads a sync target to a thread.

A WebSocket frame clones and binds a per-request agent, then streams its reply as delta frames.

sequenceDiagram
    participant Client
    participant WS as /api/chat/stream
    participant Agent

    Client->>WS: {"message": ..., "agent_name": ..., "session_id": ...}
    WS->>Agent: _isolate_agent() → clone + bind_session
    alt iter_stream and no output guardrail
        loop each chunk
            Agent-->>WS: chunk
            WS-->>Client: {"delta": "..."}
        end
    else fallback (no iter_stream, or output guardrail attached)
        alt achat exists
            WS->>Agent: await achat(message)
        else chat
            WS->>Agent: asyncio.to_thread(chat, message)
        end
        Agent-->>WS: full reply
        WS-->>Client: {"response": "..."}
    end
    WS-->>Client: {"done": true}
    Note over WS,Client: on error → {"error": "..."}; socket stays open

Endpoint reference

Three surfaces resolve under the default /api prefix (AgentOSConfig.api_prefix).

Request and response shapes

Teams and flows share one request model; the WebSocket exchanges JSON frames.

{"message": "string", "session_id": "optional-string"}

{"team": "research_team", "result": "...", "session_id": "optional-string"}

{"message": "string", "session_id": "optional-string"}

{"flow": "writer_flow", "result": "...", "session_id": "optional-string"}

// client → server
{"message": "...", "agent_name": "optional", "session_id": "optional"}

// server → client (streaming, default when the agent exposes iter_stream and has no output guardrail)
{"delta": "He"}
{"delta": "llo"}
{"done": true}

// server → client (single-frame fallback: no iter_stream, or an output guardrail is attached)
{"response": "Hello"}
{"done": true}

// server → client (error)
{"error": "..."}

Authentication (WebSocket)

The HTTP api-key middleware does not run on the WebSocket handshake. When you configure AgentOSConfig(api_key=...) or set PRAISONAI_AGENTOS_API_KEY, the WS endpoint checks the token itself and rejects unauthenticated clients with close code 1008 before accept(). Pass the token via Authorization: Bearer, X-API-Key, or ?api_key= in the URL — browsers can only use the query form. Comparison uses hmac.compare_digest.

import asyncio, json, websockets

async def main():
    url = "ws://localhost:8000/api/chat/stream?api_key=YOUR_TOKEN"
    async with websockets.connect(url) as ws:
        await ws.send(json.dumps({"message": "Hello", "agent_name": "assistant"}))
        while True:
            frame = json.loads(await ws.recv())  # {"delta": "..."} ... then {"done": true}
            print(frame)
            if frame.get("done") or "error" in frame:
                break

asyncio.run(main())

Concurrency and session isolation

Teams and flows serialise per instance; chat clones its agent per request.

• Each team and flow has a lazy per-instance asyncio.Lock — concurrent calls to the same instance queue, while different instances run in parallel (teams/flows carry mutable run state and cannot run re-entrantly).
• Per-request chat uses an isolated agent clone via _isolate_agent() (chat history is not shared across callers), including agents with handoffs. clone_for_channel tags handoffs=[...] tool closures with _praison_handoff_source and rebinds the delegation tool to the clone, so the handoff graph still works and chat_history is never shared across callers. Plain mocks and lightweight callables that don’t support isolation still fall back to the template.
• WebSocket calls use the exact same _isolate_agent() helper as POST /chat.

Choosing the right surface

Pick the surface that matches the interaction shape.

Best Practices

Related