Gateway API Endpoints

Serve OpenAI-compatible and MCP HTTP endpoints from the running gateway

Next

Turn the running gateway into an OpenAI-compatible and MCP endpoint so SDK clients and MCP tools reach the same live agents as chat users.

One process, one agent state, three protocols. This is different from praisonai serve openai, which runs a separate standalone OpenAI-only process.

Quick Start

1. Enable in gateway.yaml

   Add the api block to your gateway config:

   gateway:
     api:
       openai: true
       mcp: true

2. Start with CLI flags

   Enable the same surfaces from the command line:

   praisonai gateway start --config gateway.yaml --openai-api --mcp

3. Enable in Python

   Pass constructor flags to the gateway:

   from praisonai.gateway import WebSocketGateway
   
   WebSocketGateway(config=cfg, openai_api=True, mcp=True).start()

How It Works

Each API request dispatches into the gateway’s own registered agents, sharing the same session store and admission gate as chat users.

Endpoints Exposed

Call the OpenAI surface with any standard client:

from openai import OpenAI

client = OpenAI(base_url="http://127.0.0.1:8765/v1", api_key="gateway-token")
reply = client.chat.completions.create(
    model="assistant",
    messages=[{"role": "user", "content": "Hello"}],
)
print(reply.choices[0].message.content)

Background Runs

Submit long agent turns in the background, retrieve the result by id, and cancel in-flight runs — over the same OpenAI-compatible surface.

The synchronous path is unchanged and remains the default — omit background and store to keep the exact previous behaviour.

Which mode fits

Quick Start

1. Submit in the background

   import requests
   
   resp = requests.post(
       "http://127.0.0.1:8765/v1/responses",
       headers={"Authorization": "Bearer gateway-token"},
       json={"model": "assistant", "input": "Write me a long essay", "background": True},
   )
   run_id = resp.json()["id"]
   print(resp.status_code, resp.json()["status"])   # 202 queued

2. Poll until terminal

   import time, requests
   
   while True:
       got = requests.get(
           f"http://127.0.0.1:8765/v1/responses/{run_id}",
           headers={"Authorization": "Bearer gateway-token"},
       ).json()
       if got["status"] in ("completed", "cancelled", "failed"):
           break
       time.sleep(2)
   print(got["output_text"])

3. Cancel if needed

   import requests
   
   requests.post(
       f"http://127.0.0.1:8765/v1/responses/{run_id}/cancel",
       headers={"Authorization": "Bearer gateway-token"},
   )

User Flow

Status Vocabulary

A completed response always carries an assistant message (even with empty text), so a client reading output[0] keeps working. Polling the same completed response returns the same output[0].id each time.

Response Object

{
  "id": "resp-<hex>",
  "object": "response",
  "created_at": 1727612345,
  "model": "<agent_id>",
  "status": "queued|in_progress|completed|cancelled|failed",
  "output": [
    { "id": "msg-<hex>", "type": "message", "role": "assistant",
      "content": [{"type": "output_text", "text": "..."}] }
  ],
  "output_text": "...",
  "usage": { "input_tokens": 0, "output_tokens": 0, "total_tokens": 0 },
  "error": { "message": "..." }
}

The error block appears only when status is failed. A GET on an unknown id returns 404 with {"error": {"type": "invalid_request_error", "message": "No such response: resp-…"}}.

Both new routes go through the same _api_guarded wrapper as POST /v1/responses (gated by gateway.auth_token) and only mount when gateway.api.openai = true.

Safe Retries (Idempotency-Key)

Send the same Idempotency-Key header on a retry and the gateway returns the first run instead of starting a second — no duplicate LLM spend, no duplicate tool calls.

Opt-in, backward-compatible: no header → unchanged behaviour (every POST starts a fresh run, as before).

Quick Start

1. Send a key with /v1/responses

   Any UUID-shaped string works. Reuse the exact same value on every retry of the same logical request.

   import uuid, requests
   
   key = str(uuid.uuid4())
   body = {"model": "assistant", "input": "Write a short poem"}
   headers = {
       "Authorization": "Bearer gateway-token",
       "Idempotency-Key": key,
   }
   
   resp = requests.post("http://127.0.0.1:8765/v1/responses", json=body, headers=headers)
   print(resp.json()["output_text"])

2. Retry safely with the same key

   If the first POST timed out or your load balancer re-sent it, replay with the same key and body — you’ll get the first run’s answer back, not a second run.

   # Same key + same body → first run's response, no second agent run
   retry = requests.post("http://127.0.0.1:8765/v1/responses", json=body, headers=headers)
   assert retry.json()["id"] == resp.json()["id"]  # same run id

3. Works on /v1/chat/completions too

   import uuid, requests
   
   key = str(uuid.uuid4())
   body = {
       "model": "assistant",
       "messages": [{"role": "user", "content": "Hello"}],
   }
   headers = {"Authorization": "Bearer gateway-token", "Idempotency-Key": key}
   
   requests.post("http://127.0.0.1:8765/v1/chat/completions", json=body, headers=headers)

What happens on each case

User Flow

Scoping and limits

Token Usage

Every response reports real per-turn token counts from the agent’s LLM instance.

Each surface reports usage in its own OpenAI-native shape:

When an agent exposes no LLM metrics, usage falls back to zeros in the correct spec shape, so clients never need to special-case a missing block.

Streamed usage

Opt in with stream_options.include_usage: true to receive a streamed usage chunk — the same contract OpenAI’s own API uses.

from openai import OpenAI

client = OpenAI(base_url="http://127.0.0.1:8765/v1", api_key="gateway-token")
stream = client.chat.completions.create(
    model="assistant",
    messages=[{"role": "user", "content": "Hello"}],
    stream=True,
    stream_options={"include_usage": True},
)
for chunk in stream:
    if chunk.usage:
        print(chunk.usage)  # final chunk: real prompt/completion/total tokens

The usage-only chunk is emitted right before data: [DONE]. Without include_usage, the stream frames are unchanged and no extra chunk is sent. This is orthogonal to token-level streaming below — it works with or without it.

Token-level streaming (opt-in)

Enable gateway.api.stream to emit each DELTA_TEXT / FIRST_TOKEN piece as its own chat.completion.chunk with delta.content, so time-to-first-content matches token latency instead of full-turn latency.

Default is off: without opt-in the SSE surface is byte-for-byte the previous single buffered chunk, so existing clients keep working unchanged.

Enable it in gateway.yaml:

gateway:
  api:
    openai: true
    stream: true   # opt-in token-level SSE on /v1/chat/completions

Or in Python:

from praisonai.gateway import WebSocketGateway
from praisonaiagents.gateway import ApiConfig

WebSocketGateway(
    config=cfg,
    api=ApiConfig(openai=True, stream=True),
).start()

Consume it from any standard OpenAI client — no client-side change needed:

from openai import OpenAI

client = OpenAI(base_url="http://127.0.0.1:8765/v1", api_key="gateway-token")
stream = client.chat.completions.create(
    model="assistant",
    messages=[{"role": "user", "content": "Write a haiku about the sea"}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)  # arrives token-by-token

Behaviour guarantees:

Configuration Options

The gateway.api block maps to the ApiConfig dataclass. All three flags are opt-in; when openai and mcp are both False (default), no extra routes are mounted.

ApiConfig also exposes an enabled property (true if either surface is on), plus to_dict() and from_dict().

How Auth Works

Every API route is protected by the same gateway.auth_token as /info and /metrics.

The /info endpoint advertises which surfaces are enabled in its api field, so clients can introspect a running gateway before connecting.

GET /info also returns scopes, the caller’s effective operator scopes on this request, resolved from the bearer token against the auth_scopes policy. With no policy, every authenticated caller gets every scope.

{
  "version": "2.3.4",
  "protocol_version": 7,
  "sessions": 7,
  "clients": 3,
  "api": {},
  "scopes": ["read", "write"]
}

praisonai gateway status prints this as an Access: line — see Gateway CLI → status and Operator Scopes.

When to Use vs praisonai serve openai

Best Practices

Related