will/flynn
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
58c4b0b9bb852ad70eb8a89de5eea4c53939983d
flynn/docs/api/PROTOCOL.md
T
William Valentin 58c4b0b9bb Add iOS node push-token registration foundation
2026-02-16 12:47:34 -08:00

23 KiB
Raw Blame History

Gateway API Protocol

Flynn's gateway exposes a WebSocket-based JSON-RPC protocol for real-time communication with the AI agent. This document describes the protocol in detail.

Table of Contents

Overview

The gateway provides:

  • WebSocket Server: Real-time bidirectional communication
  • JSON-RPC 2.0: Structured request/response protocol
  • Streaming Events: Real-time updates during agent processing
  • HTTP Server: Serves static dashboard and handles webhook endpoints
  • Node Capability Negotiation: Optional companion-node role/capability registration

Execution Model (Sessions + Per-Session Queue)

Two concepts matter for correct clients:

  • connectionId: a single WebSocket connection identity (assigned on connect)
  • sessionId: the conversation/session the connection is attached to (defaults to a per-connection session, but can be switched to resume an old session)

The gateway serialises agent work per session, not per WebSocket connection:

  • Requests that target the same sessionId run one-at-a-time (FIFO) in a per-session lane.
  • Requests for different sessions can run in parallel.
  • Lane policy is configurable (collect, followup, steer_backlog, interrupt) with per-channel and per-session overrides.
  • Session-local overrides can be managed at runtime via agent.send commands: /queue, /queue set ..., /queue reset.

This is implemented via a per-lane queue (LaneQueue) in the gateway server, and used by agent.send and agent.cancel.

sequenceDiagram
  autonumber
  participant C as Client
  participant G as Gateway (WS JSON-RPC)
  participant LQ as LaneQueue (per-session)
  participant SB as SessionBridge
  participant A as AgentOrchestrator

  C->>G: agent.send {connectionId, message}
  G->>SB: resolve sessionId for connectionId
  SB-->>G: sessionId (laneId)
  G->>LQ: enqueue(laneId, work)

  alt lane idle
    LQ-->>G: starts work immediately
  else lane busy
    Note over LQ: work queued (FIFO) for this lane
  end

  G->>A: process(message) in that session
  A-->>G: streaming events (content/tool_start/tool_end)
  G-->>C: events + final done

  C->>G: agent.cancel {connectionId}
  G->>LQ: cancel(laneId) (queued items rejected)
  G->>SB: cancel active op (best-effort)
  G-->>C: result.cancelled=true/false

Base URL

  • WebSocket: ws://localhost:18800 (or wss:// if using TLS)
  • HTTP: http://localhost:18800 (or https:// if using TLS)
  • Health check: GET /health

Default Ports

  • Gateway: 18800
  • Tailscale Serve (if enabled): 443 (HTTPS)

Connection

WebSocket Handshake

// Browser
const ws = new WebSocket('ws://localhost:18800');

// Node.js (ws library)
const WebSocket = require('ws');
const ws = new WebSocket('ws://localhost:18800');

Connection Lifecycle

  1. Client connects to WebSocket endpoint
  2. Server validates authentication (if configured)
  3. Connection assigned unique ID
  4. Client can start sending requests
  5. Server sends responses and events asynchronously

Disconnection

ws.onclose = (event) => {
  console.log('Disconnected:', event.code, event.reason);
};

Common close codes:

  • 1000: Normal closure
  • 1001: Endpoint going away
  • 1006: Abnormal closure (network issue)

Authentication

Bearer Token Auth

If server.token is configured, all WebSocket connections must provide authentication:

const ws = new WebSocket('ws://localhost:18800', {
  headers: {
    'Authorization': 'Bearer your-secret-token'
  }
});

Tailscale Identity

If server.tailscale_identity is enabled, connections from Tailscale are trusted based on the Tailscale-User-Login header.

// Automatic when connecting via Tailscale
// No additional auth required if server.tailscale_identity is enabled

HTTP Auth

If server.auth_http is true (default: true), HTTP requests also require bearer token when server.token is set:

fetch('http://localhost:18800/api/health', {
  headers: {
    'Authorization': 'Bearer your-secret-token'
  }
});

Exceptions (handled by their own trust/auth model and therefore bypass gateway token auth):

  • POST /webhooks/:name (HMAC-validated when webhook secret is configured)
  • POST /gmail/push (Google Pub/Sub push)
  • POST /teams/events (Microsoft Bot Framework activity callback)
  • POST /google-chat/events (Google Chat event callback, optional webhook token check)
  • POST /bluebubbles/events (BlueBubbles iMessage webhook callback, optional webhook token check)

Message Format

Request (Client → Server)

interface GatewayRequest {
  id: number;           // Unique request ID (integer)
  method: string;       // Method name (e.g., 'agent.send', 'system.info')
  params?: Record<string, unknown>;  // Method parameters (optional)
}

Example:

{
  "id": 1,
  "method": "agent.send",
  "params": {
    "message": "Hello, Flynn!",
    "session": "telegram:123456"
  }
}

Response (Server → Client)

interface GatewayResponse {
  id: number;           // Request ID this responds to
  result: unknown;      // Method result
}

Example:

{
  "id": 1,
  "result": {
    "sessionId": "telegram:123456",
    "response": "Hello! How can I help you today?"
  }
}

Error (Server → Client)

interface GatewayError {
  id: number;
  error: {
    code: ErrorCode;     // Error code (integer)
    message: string;     // Human-readable error message
  };
}

Example:

{
  "id": 1,
  "error": {
    "code": 4,
    "message": "Request cancelled by client"
  }
}

Event (Server → Client)

interface GatewayEvent {
  id: number;           // Request ID this relates to
  event: EventType;      // Event type string
  data: unknown;        // Event-specific data
}

Example:

{
  "id": 1,
  "event": "content",
  "data": {
    "text": "Hello! How can I help you today?"
  }
}

Methods

System Methods

system.info

Get gateway information.

Request:

{
  "id": 1,
  "method": "system.info"
}

Response:

{
  "id": 1,
  "result": {
    "version": "0.1.0",
    "uptime": 12345,
    "connections": 2
  }
}

system.disconnect

Close the connection gracefully.

Request:

{
  "id": 2,
  "method": "system.disconnect"
}

system.presence

Return tracked sender presence snapshots (most recent first).

Online/offline is inferred from inactivity threshold in the daemon.

Request:

{
  "id": 3,
  "method": "system.presence",
  "params": {
    "channel": "telegram",
    "status": "online",
    "limit": 50
  }
}

Response:

{
  "id": 3,
  "result": {
    "presence": [
      {
        "channel": "telegram",
        "senderId": "123456",
        "senderName": "alice",
        "firstSeenAt": 1739700000000,
        "lastSeenAt": 1739700300000,
        "messageCount": 12,
        "status": "online"
      }
    ],
    "summary": { "total": 1, "online": 1, "offline": 0 }
  }
}

Response:

{
  "id": 2,
  "result": {
    "success": true
  }
}

Session Methods

sessions.list

List all sessions.

Request:

{
  "id": 3,
  "method": "sessions.list"
}

Response:

{
  "id": 3,
  "result": {
    "sessions": [
      {
        "id": "telegram:123456",
        "createdAt": "2025-02-13T10:00:00Z",
        "lastActiveAt": "2025-02-13T12:00:00Z",
        "messageCount": 42,
        "connectionCount": 1
      }
    ]
  }
}

sessions.get

Get session details.

Request:

{
  "id": 4,
  "method": "sessions.get",
  "params": {
    "sessionId": "telegram:123456"
  }
}

Response:

{
  "id": 4,
  "result": {
    "id": "telegram:123456",
    "createdAt": "2025-02-13T10:00:00Z",
    "lastActiveAt": "2025-02-13T12:00:00Z",
    "history": [
      {
        "role": "user",
        "content": "Hello"
      },
      {
        "role": "assistant",
        "content": "Hi there!"
      }
    ]
  }
}

sessions.create

Create or resume a session.

Request:

{
  "id": 5,
  "method": "sessions.create",
  "params": {
    "sessionId": "telegram:123456"
  }
}

Response:

{
  "id": 5,
  "result": {
    "sessionId": "telegram:123456",
    "created": true
  }
}

sessions.delete

Delete a session.

Request:

{
  "id": 6,
  "method": "sessions.delete",
  "params": {
    "sessionId": "telegram:123456"
  }
}

Response:

{
  "id": 6,
  "result": {
    "success": true
  }
}

Agent Methods

agent.send

Send a message to the agent and stream response.

Request:

{
  "id": 7,
  "method": "agent.send",
  "params": {
    "message": "What's the weather?",
    "sessionId": "telegram:123456",
    "attachments": [
      {
        "mimeType": "image/jpeg",
        "data": "base64encodedimage..."
      }
    ]
  }
}

Response (final):

{
  "id": 7,
  "result": {
    "content": "I can't check the weather without access to weather APIs. Would you like me to help you with something else?",
    "usage": {
      "inputTokens": 25,
      "outputTokens": 30
    }
  }
}

Events (streamed during processing):

content event:

{
  "id": 7,
  "event": "content",
  "data": {
    "text": "I can't check the weather"
  }
}

tool_start event:

{
  "id": 7,
  "event": "tool_start",
  "data": {
    "tool": "shell.exec",
    "args": {
      "command": "echo 'Hello'"
    }
  }
}

tool_end event:

{
  "id": 7,
  "event": "tool_end",
  "data": {
    "tool": "shell.exec",
    "result": {
      "success": true,
      "output": "Hello\n",
      "error": null
    }
  }
}

attachment event:

{
  "id": 7,
  "event": "attachment",
  "data": {
    "mimeType": "image/png",
    "data": "base64encoded..."
  }
}

done event:

{
  "id": 7,
  "event": "done",
  "data": {
    "content": "Complete response here..."
  }
}

When queue policy rejects/supersedes a request before execution, the server emits an error event with code: 3 (AgentBusy) and includes data.queue metadata (code, laneId, mode, overflow, droppedCount).

agent.cancel

Cancel the current agent operation.

Node Methods

node.register

Register node role/capabilities for the current WebSocket connection.

Request:

{
  "id": 9,
  "method": "node.register",
  "params": {
    "nodeId": "companion-desktop",
    "role": "companion",
    "protocolVersion": 1,
    "capabilities": ["ui.canvas", "notifications"]
  }
}

Response:

{
  "id": 9,
  "result": {
    "registered": true,
    "node": { "id": "companion-desktop", "role": "companion" },
    "protocol": { "serverVersion": 1, "clientVersion": 1, "negotiatedVersion": 1 },
    "capabilities": {
      "declared": ["ui.canvas", "notifications"],
      "enabled": ["ui.canvas", "notifications"]
    }
  }
}

node.capabilities.get

Return negotiated capabilities for the currently registered node connection.

node.location.set

Update the last-known location for the currently registered node connection. Requires server.nodes.enabled: true and server.nodes.location.enabled: true.

Request:

{
  "id": 10,
  "method": "node.location.set",
  "params": {
    "latitude": 37.7749,
    "longitude": -122.4194,
    "accuracyMeters": 12.4,
    "source": "gps",
    "capturedAt": 1763241200000
  }
}

Response:

{
  "id": 10,
  "result": {
    "updated": true,
    "node": { "id": "companion-desktop", "role": "companion" },
    "location": {
      "latitude": 37.7749,
      "longitude": -122.4194,
      "accuracyMeters": 12.4,
      "source": "gps",
      "capturedAt": 1763241200000,
      "receivedAt": 1763241200451
    }
  }
}

node.location.get

Return the stored last-known location for the currently registered node connection.

node.status.set

Publish companion/node runtime status metadata (for example macOS menu-bar heartbeat state).

Request:

{
  "id": 12,
  "method": "node.status.set",
  "params": {
    "platform": "macos",
    "appVersion": "0.3.1",
    "deviceName": "MacBook Pro",
    "statusText": "Idle",
    "batteryPct": 64,
    "powerSource": "battery"
  }
}

node.push_token.set

Register a node push token (currently APNs) for companion delivery routing. Requires server.nodes.push.enabled: true.

Request:

{
  "id": 13,
  "method": "node.push_token.set",
  "params": {
    "provider": "apns",
    "token": "abcd1234abcd1234abcd1234abcd1234",
    "topic": "com.example.flynn",
    "environment": "sandbox"
  }
}

system.capabilities

Return gateway protocol version, node policy status, and feature-gate snapshot.

Request:

{
  "id": 11,
  "method": "system.capabilities"
}

Response:

{
  "id": 11,
  "result": {
    "protocol": { "version": 1 },
    "nodes": {
      "enabled": true,
      "locationEnabled": true,
      "pushEnabled": true,
      "allowedRoles": ["companion"],
      "registered": true,
      "role": "companion",
      "nodeId": "companion-desktop"
    },
    "featureGates": {
      "ui.canvas": true
    }
  }
}

system.location

Return the operator-facing snapshot of registered node locations.

system.nodes

Return the operator-facing snapshot of registered node connections (identity, role, capabilities, location/status). Push tokens are returned as masked previews (tokenPreview) and never exposed in full.

Canvas Methods

canvas.put

Upsert a session-scoped canvas artifact.

Request:

{
  "id": 12,
  "method": "canvas.put",
  "params": {
    "sessionId": "ws:abc123",
    "artifactId": "summary-card",
    "type": "note",
    "title": "Draft Summary",
    "content": { "markdown": "## Notes" },
    "metadata": { "lane": "analysis" }
  }
}

canvas.get

Fetch a single artifact by id.

canvas.list

List artifacts for a session (newest first).

canvas.delete

Delete a single artifact.

canvas.clear

Delete all artifacts for a session.

agent.setToolUseCallback

Set callback for tool use events (for confirmation UI).

Request:

{
  "id": 9,
  "method": "agent.setToolUseCallback",
  "params": {
    "sessionId": "telegram:123456",
    "enabled": true
  }
}

Response:

{
  "id": 9,
  "result": {
    "success": true
  }
}

Tool Methods

tools.list

List available tools.

Request:

{
  "id": 10,
  "method": "tools.list",
  "params": {
    "sessionId": "telegram:123456"
  }
}

Response:

{
  "id": 10,
  "result": {
    "tools": [
      {
        "name": "shell.exec",
        "description": "Execute a shell command...",
        "inputSchema": {
          "type": "object",
          "properties": {
            "command": { "type": "string" }
          }
        }
      }
    ]
  }
}

tools.execute

Execute a tool directly (bypass agent).

Request:

{
  "id": 11,
  "method": "tools.execute",
  "params": {
    "sessionId": "telegram:123456",
    "tool": "shell.exec",
    "args": {
      "command": "echo 'Hello'"
    }
  }
}

Response:

{
  "id": 11,
  "result": {
    "success": true,
    "output": "Hello\n"
  }
}

Config Methods

config.get

Get current configuration (secrets redacted).

Request:

{
  "id": 12,
  "method": "config.get"
}

Response:

{
  "id": 12,
  "result": {
    "models": {
      "default": {
        "anthropic": {
          "apiKey": "***"
        }
      }
    }
  }
}

config.reload

Reload configuration from file.

Request:

{
  "id": 13,
  "method": "config.reload"
}

Response:

{
  "id": 13,
  "result": {
    "success": true
  }
}

Pairing Methods

pairing.generate

Generate a pairing code for unknown senders.

Request:

{
  "id": 14,
  "method": "pairing.generate",
  "params": {
    "channel": "telegram"
  }
}

Response:

{
  "id": 14,
  "result": {
    "code": "ABC123",
    "expiresAt": "2025-02-13T12:05:00Z"
  }
}

pairing.list

List active pairing codes.

Request:

{
  "id": 15,
  "method": "pairing.list"
}

Response:

{
  "id": 15,
  "result": {
    "codes": [
      {
        "code": "ABC123",
        "channel": "telegram",
        "createdAt": "2025-02-13T12:00:00Z",
        "expiresAt": "2025-02-13T12:05:00Z"
      }
    ]
  }
}

Events

Event Types

content

Streamed text content from the agent.

{
  "id": 1,
  "event": "content",
  "data": {
    "text": "Partial response..."
  }
}

tool_start

Tool execution started.

{
  "id": 1,
  "event": "tool_start",
  "data": {
    "tool": "shell.exec",
    "args": {
      "command": "echo 'test'"
    }
  }
}

tool_end

Tool execution completed.

{
  "id": 1,
  "event": "tool_end",
  "data": {
    "tool": "shell.exec",
    "result": {
      "success": true,
      "output": "test\n"
    }
  }
}

attachment

Outbound attachment (image, audio, file).

{
  "id": 1,
  "event": "attachment",
  "data": {
    "mimeType": "image/jpeg",
    "data": "base64encoded...",
    "filename": "output.jpg"
  }
}

done

Agent processing complete (final response).

{
  "id": 1,
  "event": "done",
  "data": {
    "content": "Complete final response..."
  }
}

error

Error occurred during processing.

{
  "id": 1,
  "event": "error",
  "data": {
    "code": 5,
    "message": "Internal error: ...",
    "queue": {
      "code": "overflow",
      "laneId": "ws:abc123",
      "mode": "followup",
      "overflow": "drop_new",
      "droppedCount": 1
    }
  }
}

data.queue is optional and only present for queue policy rejections/superseded requests.

Error Codes

Code Name Description
-1 ParseError Invalid JSON in request
-2 InvalidRequest Missing required fields
-3 MethodNotFound Unknown method name
-4 AuthRequired Authentication required but not provided
-5 AuthFailed Authentication failed
1 SessionNotFound Session ID doesn't exist
2 ToolNotFound Tool name doesn't exist
3 AgentBusy Agent is processing another request
4 RequestCancelled Request was cancelled by client
5 InternalError Unexpected server error

Example Client

Browser Client

class FlynnClient {
  constructor(url, token) {
    this.url = url;
    this.token = token;
    this.requestId = 0;
    this.pending = new Map();
  }

  connect() {
    const headers = {};
    if (this.token) {
      headers['Authorization'] = `Bearer ${this.token}`;
    }

    this.ws = new WebSocket(this.url, { headers });

    this.ws.onopen = () => {
      console.log('Connected to Flynn gateway');
    };

    this.ws.onmessage = (event) => {
      const message = JSON.parse(event.data);
      this.handleMessage(message);
    };

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };

    this.ws.onclose = () => {
      console.log('Disconnected from Flynn gateway');
    };
  }

  sendRequest(method, params = {}) {
    return new Promise((resolve, reject) => {
      const id = ++this.requestId;

      const request = {
        id,
        method,
        params
      };

      this.ws.send(JSON.stringify(request));

      // Store promise for response
      this.pending.set(id, { resolve, reject });

      // Set timeout
      setTimeout(() => {
        if (this.pending.has(id)) {
          this.pending.delete(id);
          reject(new Error('Request timeout'));
        }
      }, 30000);
    });
  }

  handleMessage(message) {
    // Response
    if ('result' in message) {
      const { id, result } = message;
      const pending = this.pending.get(id);
      if (pending) {
        this.pending.delete(id);
        pending.resolve(result);
      }
    }

    // Error
    else if ('error' in message) {
      const { id, error } = message;
      const pending = this.pending.get(id);
      if (pending) {
        this.pending.delete(id);
        const err = new Error(error.message);
        err.code = error.code;
        pending.reject(err);
      }
    }

    // Event
    else if ('event' in message) {
      this.handleEvent(message);
    }
  }

  handleEvent(event) {
    const { id, event: eventType, data } = event;

    switch (eventType) {
      case 'content':
        console.log('Content:', data.text);
        break;
      case 'tool_start':
        console.log('Tool started:', data.tool, data.args);
        break;
      case 'tool_end':
        console.log('Tool completed:', data.tool, data.result);
        break;
      case 'attachment':
        console.log('Attachment received:', data.mimeType);
        break;
      case 'done':
        console.log('Done:', data.content);
        break;
      case 'error':
        console.error('Error:', data.code, data.message);
        break;
    }
  }

  // Convenience methods
  async systemInfo() {
    return this.sendRequest('system.info');
  }

  async listSessions() {
    return this.sendRequest('sessions.list');
  }

  async sendMessage(message, sessionId, attachments = []) {
    return this.sendRequest('agent.send', {
      message,
      sessionId,
      attachments
    });
  }

  async listTools(sessionId) {
    return this.sendRequest('tools.list', { sessionId });
  }
}

// Usage
const client = new FlynnClient('ws://localhost:18800', 'your-token');
client.connect();

client.onopen = async () => {
  const info = await client.systemInfo();
  console.log('Gateway info:', info);

  const response = await client.sendMessage('Hello, Flynn!', 'telegram:123456');
  console.log('Response:', response);
};

Node.js Client

const WebSocket = require('ws');

class FlynnNodeClient extends FlynnClient {
  connect() {
    const options = {};
    if (this.token) {
      options.headers = { 'Authorization': `Bearer ${this.token}` };
    }

    this.ws = new WebSocket(this.url, options);

    // ... same as browser client
  }
}

// Usage
const client = new FlynnNodeClient('ws://localhost:18800', 'your-token');
client.connect();

HTTP Fetch Example

// Health check
async function checkHealth() {
  const response = await fetch('http://localhost:18800/health', {
    headers: {
      'Authorization': 'Bearer your-token'
    }
  });

  const status = await response.json();
  console.log('Health:', status);
}

checkHealth();

For more implementation details, see:

  • Protocol types: src/gateway/protocol.ts
  • Handlers: src/gateway/handlers/
  • Gateway server: src/gateway/server.ts
Reference in New Issue View Git Blame Copy Permalink