> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
    Client[📋 Client] --> OS[🧠 AgentOS]
    OS --> Teams[🤖 POST /api/teams/name/run]
    OS --> Flows[🤖 POST /api/flows/name/run]
    OS --> WS[💬 WS /api/chat/stream]

    classDef client fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef os fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef surface fill:#8B0000,stroke:#7C90A0,color:#fff

    class Client client
    class OS os
    class Teams,Flows,WS surface
```

## 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.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.

```bash theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
    participant Client
    participant AgentOS
    participant Target as Team / Flow

    Client->>AgentOS: POST /api/teams/{name}/run
    AgentOS->>AgentOS: look up name in self.teams (404 if unknown)
    AgentOS->>Target: acquire per-instance asyncio.Lock
    alt arun / astart exists
        AgentOS->>Target: await arun / astart
    else only run / start
        AgentOS->>Target: asyncio.to_thread(run / start)
    end
    Target-->>AgentOS: result
    AgentOS-->>Client: {"team": name, "result": ..., "session_id": ...}
```

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

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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`).

| 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:
>
> ```json theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
> {"message": "string", "session_id": "optional-string"}
> ```
>
> Response:
>
> ```json theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
> {"team": "research_team", "result": "...", "session_id": "optional-string"}
> ```

> **POST /api/flows/\{flow\_name}/run**
>
> Same shape as teams, with `flow` instead of `team`.
>
> Request:
>
> ```json theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
> {"message": "string", "session_id": "optional-string"}
> ```
>
> Response:
>
> ```json theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
> {"flow": "writer_flow", "result": "...", "session_id": "optional-string"}
> ```

> **WS /api/chat/stream frames**
>
> ```json theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
> // 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`.

```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.

```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
    Q{What are you invoking?} -->|Single-turn chat| C[POST /api/chat]
    Q -->|Multi-step team| T[POST /api/teams/name/run]
    Q -->|Structured pipeline| F[POST /api/flows/name/run]
    Q -->|Interactive turn-by-turn| W[WS /api/chat/stream]

    classDef question fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef answer fill:#10B981,stroke:#7C90A0,color:#fff

    class Q question
    class C,T,F,W answer
```

***

## 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](https://praison.ai/docs/features/agentos-read-endpoints.md)**
>
> Health, agents, approvals, and other read routes.

> **[AgentOS Serve Reload](https://praison.ai/docs/features/agentos-serve-reload.md)**
>
> Hot-reload agents without restarting the server.

> **[Session Persistence](https://praison.ai/docs/features/session-persistence.md)**
>
> Carry conversation state across calls with `session_id`.

> **[Approval Protocol](https://praison.ai/docs/features/approval-protocol.md)**
>
> Gate risky tool calls behind an approval registry.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.