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.

---

Endpoint reference

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

Method	Path	Purpose	Error contract
POST	/api/teams/{team_name}/run	Invoke a team by name.	404 if team_name doesn’t match a getattr(team, "name", None) in self.teams; 501 if the team exposes no arun/astart/run/start; 500 on team error.
POST	/api/flows/{flow_name}/run	Invoke a flow by name.	Same as teams but checks self.flows.
WS	/api/chat/stream	Chat with an agent over WebSocket.	Closes with code 1008 if a configured launch token is missing/invalid; sends {"error": "..."} on bad frames and keeps the socket open.

---

Request and response shapes

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

POST /api/teams/{team_name}/run

Request:

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

Response:

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

POST /api/flows/{flow_name}/run

Same shape as teams, with flow instead of team.

Request:

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

Response:

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

WS /api/chat/stream frames

// 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

Give every team and flow a stable name

The lookup matches getattr(team, "name", None) against the path. Without a name, the call silently returns 404.

Attach a launch token in production

Set AgentOSConfig(api_key=...) (or PRAISONAI_AGENTOS_API_KEY) so the WebSocket endpoint isn’t an unauthenticated back door — the HTTP api-key middleware does not cover the WS handshake.

Treat WebSocket errors as sticky

The socket stays open on {"error": ...}, so a retry can send another frame without a new handshake.

An agent with an output guardrail streams as one frame

Streaming yields tokens before the full response exists, so output guardrails cannot validate chunks. When agent.guardrail is set, /api/chat/stream falls back to {"response": ...} followed by {"done": true} to preserve validation.

When the client disconnects, the server stops the stream

The worker that pumps iter_stream is signalled to close as soon as the WebSocket is gone, so a stalled model or tool call does not pin the handler.

Construct one instance per concurrent caller

Teams and flows serialise same-instance calls. For true concurrency, build multiple team/flow instances — one per concurrent caller.

---

Related

AgentOS Read Endpoints

Health, agents, approvals, and other read routes.

AgentOS Serve Reload

Hot-reload agents without restarting the server.

Session Persistence

Carry conversation state across calls with session_id.

Approval Protocol

Gate risky tool calls behind an approval registry.