Gateway Operator Scopes

Least-privilege role-based access control for multi-operator Gateway deployments

Operator scopes grant teammates least-privilege access to a shared Gateway — read-only dashboards, send-but-not-approve operators, or full admins — without handing over the whole keys.

from praisonaiagents import Agent

agent = Agent(
    name="assistant",
    instructions="You are a helpful assistant.",
)
agent.start("Send a message through the gateway")

The user assigns scoped roles; each operator reaches only the gateway actions their role allows.

Quick Start

1. Single-operator (no scopes — unchanged)

   Today’s setup keeps working. Authenticated clients receive all scopes when no policy is configured.

   from praisonaiagents import Agent
   
   agent = Agent(
       name="assistant",
       instructions="You are a helpful assistant.",
   )
   
   # $ praisonai gateway start --host 127.0.0.1
   agent.start("hello")

2. Multi-operator (scoped tokens)

   Map each operator token to the scopes they need in gateway.yaml, then run your agent as usual.

   gateway:
     host: "0.0.0.0"
     port: 8765
     auth:
       tokens:
         - token: "${VIEWER_TOKEN}"
           scopes: [read]
         - token: "${OPS_TOKEN}"
           scopes: [read, write]
         - token: "${ADMIN_TOKEN}"
           scopes: [admin]
   
   agents:
     assistant:
       instructions: "You are a helpful assistant."
       model: gpt-4o-mini

   from praisonaiagents import Agent
   from praisonaiagents.gateway import OperatorScope
   
   agent = Agent(name="assistant", instructions="You are a helpful assistant.")
   # OperatorScope.READ, .WRITE, .APPROVALS, .PAIRING, .ADMIN
   print([s.value for s in OperatorScope.all()])

How It Works

1. Client connects with a bearer token.
2. Gateway resolves scopes via GatewayConfig.resolve_scopes(token).
3. Each HTTP route and WebSocket action checks the required scope.
4. Outbound events are filtered — approval events only reach clients with the approvals scope.

WebSocket frames go through the central authorize_method gate before any handler runs.

Scope Reference

Which scope should this operator have?

Per-user attribution under a shared token

Operators sharing one gateway token collapse to a single audit identity by default. A proxy or SSO layer can stamp the X-Operator-Id header per user so each decision is still attributed to an individual on the audit trail.

curl -X POST -H "Authorization: Bearer $OPS_TOKEN" \
  -H "X-Operator-Id: alice" \
  -H "Content-Type: application/json" \
  -d '{"request_id": "apr-1a2b3c4d", "approved": true}' \
  http://localhost:8080/approval/resolve

The gateway stamps this as operator:alice on the audit trail. See Approval Attribution for the full identity derivation order.

Configuration

YAML — structured (recommended)

gateway:
  auth:
    tokens:
      - token: "${VIEWER_TOKEN}"
        scopes: [read]
      - token: "${OPS_TOKEN}"
        scopes: [read, write, approvals]
      - token: "${ADMIN_TOKEN}"
        scopes: [admin]

YAML — flat mapping

gateway:
  auth_scopes:
    "${VIEWER_TOKEN}": [read]
    "${OPS_TOKEN}": [read, write, approvals]
    "${ADMIN_TOKEN}": [admin]

Python

from praisonaiagents.gateway import GatewayConfig, OperatorScope

config = GatewayConfig(
    host="0.0.0.0",
    port=8765,
    auth_token="${ADMIN_TOKEN}",
    auth_scopes={
        "${VIEWER_TOKEN}": [OperatorScope.READ.value],
        "${OPS_TOKEN}": [OperatorScope.READ.value, OperatorScope.WRITE.value],
        "${ADMIN_TOKEN}": [OperatorScope.ADMIN.value],
    },
)

print(config.has_scope_policy)  # True when auth_scopes is non-empty
print(config.resolve_scopes("${VIEWER_TOKEN}"))  # ['read']

Confining operators to specific agents (Issue #6253)

On a multi-agent gateway, an operator token can be confined to a subset of agents by giving it {scopes, agents} instead of a plain scope list.

Quick Start

1. Flat auth_scopes (recommended)

   gateway:
     host: 0.0.0.0
     port: 8765
     auth_scopes:
       "${SUPPORT_TOKEN}":
         scopes: [read, write]
         agents: [support]
       "${FINANCE_TOKEN}":
         scopes: [read, write]
         agents: [finance]
       "${ADMIN_TOKEN}": [admin]        # unchanged — reaches every agent
   
   agents:
     support:  { instructions: "..." }
     finance:  { instructions: "..." }

2. Structured auth.tokens[] (equivalent)

   gateway:
     auth:
       tokens:
         - token: "${SUPPORT_TOKEN}"
           scopes: [read, write]
           agents: [support]
         - token: "${ADMIN_TOKEN}"
           scopes: [admin]              # no `agents:` → unconfined

3. Python

   import os
   from praisonaiagents.gateway import GatewayConfig, OperatorScope
   
   cfg = GatewayConfig(
       host="0.0.0.0",
       port=8765,
       auth_token=os.environ["ADMIN_TOKEN"],
       auth_scopes={
           os.environ["SUPPORT_TOKEN"]: {
               "scopes": [OperatorScope.READ.value, OperatorScope.WRITE.value],
               "agents": ["support"],
           },
           os.environ["ADMIN_TOKEN"]: [OperatorScope.ADMIN.value],  # unconfined
       },
   )

How it works

A confined operator is refused wherever a session would bind to, resume, join, or receive events from an agent outside its grant:

• hello / join bind — an agent_id outside the grant is refused with AGENT_NOT_FOUND.
• join session resume — a session_id owned by an out-of-grant agent is refused (Session does not belong to the requested agent) before the session is resumed; sessions rehydrated from the store are checked again after resume.
• attach() (shared sessions) — raises PermissionError for a session whose agent is outside the grant.
• broadcast() — events whose source is an out-of-grant agent are withheld. System (non-agent) events are unaffected.

Fail-closed guards

Python API

GatewayConfig.resolve_agents(token) returns the agent ids a token is confined to, or None when the token is unconfined (or unknown).

from praisonaiagents.gateway import GatewayConfig

cfg = GatewayConfig(
    auth_token="tmp",
    auth_scopes={
        "tok_support": {"scopes": ["read", "write"], "agents": ["support"]},
        "tok_legacy":  ["read", "write"],
    },
)

cfg.resolve_scopes("tok_support")   # ['read', 'write']
cfg.resolve_agents("tok_support")   # ['support']  — confined
cfg.resolve_agents("tok_legacy")    # None         — unconfined (every agent)
cfg.resolve_agents("nope")          # None

Scope-Gated Routes

For programmatic (WebSocket / RPC) methods, see Method → Scope Registry below.

The WebSocket message requirement comes from the Method Scope Registry below — the dispatcher calls resolve_required_scope("message") (and resolve_required_scope("agent.message")) rather than hard-coding the check per route.

Method Scope Registry

New gateway methods are closed until classified — you cannot accidentally ship an unauthenticated control-plane call.

Every method maps to a GatewayMethodDescriptor in the module-level GATEWAY_METHODS registry. resolve_required_scope looks up that descriptor and returns the effective scope for a call.

Quick lookup

from praisonaiagents import Agent
from praisonaiagents.gateway import resolve_required_scope, OperatorScope

agent = Agent(name="assistant", instructions="You are a helpful assistant.")

print(resolve_required_scope("agent.message"))   # OperatorScope.WRITE
print(resolve_required_scope("session.status"))  # OperatorScope.READ
print(resolve_required_scope("myplugin.unknown"))  # OperatorScope.ADMIN (default-deny)

Registering a plugin method

Plugin authors adding a new dispatcher method register its required scope once, at import time.

from praisonaiagents import Agent
from praisonaiagents.gateway import register_gateway_method, OperatorScope

agent = Agent(name="assistant", instructions="You are a helpful assistant.")

register_gateway_method(
    "myplugin.reindex",
    scope=OperatorScope.ADMIN,
    owner="myplugin",
)

register_gateway_method refuses duplicates — pass replace=True to override an existing entry.

Per-field escalation

Escalation raises the required scope when a sensitive field is present; a plain call stays at the baseline.

from praisonaiagents import Agent
from praisonaiagents.gateway import register_gateway_method, resolve_required_scope, OperatorScope

agent = Agent(name="assistant", instructions="You are a helpful assistant.")

register_gateway_method(
    "myplugin.publish",
    scope=OperatorScope.WRITE,
    owner="myplugin",
    escalate_fields={"config": OperatorScope.ADMIN},
    replace=True,
)

print(resolve_required_scope("myplugin.publish", {"text": "hi"}))       # OperatorScope.WRITE
print(resolve_required_scope("myplugin.publish", {"config": {...}}))    # OperatorScope.ADMIN

Escalation only ever raises the requirement — a field mapped to a weaker scope than the baseline is ignored.

Strict fields (fail-closed on unknown payload)

strict_fields=True escalates any structural field not explicitly listed as safe, so unexpected payload keys fail closed.

from praisonaiagents import Agent
from praisonaiagents.gateway import register_gateway_method, resolve_required_scope, OperatorScope

agent = Agent(name="assistant", instructions="You are a helpful assistant.")

register_gateway_method(
    "myplugin.echo",
    scope=OperatorScope.WRITE,
    owner="myplugin",
    strict_fields=True,
    safe_fields={"text"},
    replace=True,
)

print(resolve_required_scope("myplugin.echo", {"text": "hi"}))              # OperatorScope.WRITE
print(resolve_required_scope("myplugin.echo", {"text": "hi", "raw": 1}))    # OperatorScope.ADMIN

Core method classification

Core methods are classified at import; anything unlisted defaults to admin.

Connection-lifecycle frames (hello, join, leave) run before a client has negotiated any scope, so they classify as read — the lowest meaningful level. abort (and its wire alias message_abort) mutates the turn just like sending one, so both carry the same write requirement as message.

server_reply is unique — the handler escalates to approvals for approval-kind replies (parity with approvals.resolve), so a WRITE-only role can answer choice/input but not approval prompts. See Gateway Server→Client Request Channel.

Central authorization gate

Every gateway method now flows through one default-deny gate before any handler runs.

Nothing changes for end users — the gateway gates dispatch for you.

from praisonaiagents import Agent

agent = Agent(name="assistant", instructions="You are a helpful assistant.")
agent.start("hello")  # dispatch is gated automatically; nothing to configure

Plugin authors extending the dispatcher can call the gate directly.

from praisonaiagents.gateway import (
    authorize_method,
    GatewayUnauthorized,
    OperatorScope,
)

try:
    authorize_method("message", {OperatorScope.READ})
except GatewayUnauthorized as exc:
    print(exc.method, exc.required.value)
    # -> message write

• Called once, centrally, at the top of _handle_client_message — no per-handler wiring needed.
• Default-deny: an unclassified method requires admin.
• admin implies every scope.
• read is the baseline lifecycle scope — any operator holding at least one actionable scope (write / approvals / pairing / admin) satisfies read, so a write-only operator can still complete hello / join / leave / session.status. An empty scope set cannot satisfy even read.
• A malformed method (non-string / unhashable) fails closed to admin rather than raising TypeError.
• Transport PING / PONG frames bypass the gate (handled earlier in the wire loop).

Method → Scope Registry

Every gateway method declares its required scope in one place — a declarative registry — instead of scattered checks per endpoint. New methods are closed until explicitly classified; an unregistered method requires admin.

from praisonaiagents.gateway import (
    OperatorScope,
    register_gateway_method,
    resolve_required_scope,
)

# What scope does this call need?
resolve_required_scope("agent.message")            # OperatorScope.WRITE
resolve_required_scope("channels.pause")           # OperatorScope.ADMIN
resolve_required_scope("totally.unknown.method")   # OperatorScope.ADMIN (default-deny)

Core method classification

Core methods are classified at import in praisonaiagents.gateway.protocols:

Connection-lifecycle frames (hello, join, leave) run before a client has negotiated any scope, so they classify as read — the lowest meaningful level. abort (and its wire alias message_abort) mutates the turn just like sending one, so both carry the same write requirement as message.

Registering a plugin method

A plugin that adds a new gateway surface registers its scope at import time:

from praisonaiagents.gateway import OperatorScope, register_gateway_method

register_gateway_method(
    "myplugin.notify",
    scope=OperatorScope.WRITE,
    owner="myplugin",
    since="1.4.0",
)

Per-payload-field escalation (raise, never lower)

A descriptor can raise the required scope when a specific field appears in the params — an otherwise write method that mutates configuration should require admin when the config field is present:

from praisonaiagents.gateway import OperatorScope, register_gateway_method, resolve_required_scope

register_gateway_method(
    "myplugin.update",
    scope=OperatorScope.WRITE,
    escalate_fields={"config": OperatorScope.ADMIN},
)

resolve_required_scope("myplugin.update", {"text": "hi"})               # WRITE
resolve_required_scope("myplugin.update", {"text": "hi", "config": {}}) # ADMIN

Strict-fields mode (fail closed on unknown fields)

For high-risk methods, enable strict_fields=True with an allowlist of known-safe fields. Any field not in safe_fields or escalate_fields escalates to admin:

from praisonaiagents.gateway import OperatorScope, register_gateway_method, resolve_required_scope

register_gateway_method(
    "myplugin.execute",
    scope=OperatorScope.WRITE,
    strict_fields=True,
    safe_fields={"text"},
)

resolve_required_scope("myplugin.execute", {"text": "hi"})              # WRITE
resolve_required_scope("myplugin.execute", {"text": "hi", "mutate": 1}) # ADMIN

Incomparable scopes escalate to ADMIN

approvals and pairing are sibling capabilities — neither implies the other. When a method’s baseline is one and a field escalates to the other, the combined requirement escalates to admin rather than silently picking one (a single-scope check cannot satisfy both):

from praisonaiagents.gateway import OperatorScope, register_gateway_method, resolve_required_scope

register_gateway_method(
    "myplugin.combo",
    scope=OperatorScope.APPROVALS,
    escalate_fields={"pair": OperatorScope.PAIRING},
)

resolve_required_scope("myplugin.combo", {"other": 1})   # APPROVALS
resolve_required_scope("myplugin.combo", {"pair": True}) # ADMIN — fail-closed

API reference

For the approval-flow context these scopes gate, see Scoped Approvals.

Common Patterns

Read-only dashboard viewer — [read] for status and transcripts without send or approve rights.

Send but not approve — [read, write] for operators who reply to users but cannot resolve tool approvals.

Approvals-only on-call — [read, approvals] for security-sensitive approval resolution without channel admin rights.

Full admin — [admin] for SREs who need pause/resume/reconnect plus all other capabilities.

Introspecting your own scopes

praisonai gateway status prints the scopes your own token holds on a running gateway, no curl needed.

$ praisonai gateway status
...
  Access: authorized (scopes: read)

When the token has neither write nor admin, status also prints a Next step: hint pointing at gateway.auth_scopes. The same data is the scopes field of GET /info; see Gateway CLI → status for all verdicts.

Error Handling

HTTP 403 when scope check fails:

{ "error": "insufficient scope", "required_scope": "approvals" }

WebSocket message without write scope:

{
  "type": "error",
  "code": "insufficient_scope",
  "message": "insufficient scope",
  "required_scope": "write"
}

Best Practices

Related