relay/reducer

The pure core of an MCP server: a reducer from inputs to effects, for authors of custom transports and runtimes.

init(server) returns the State of one connection. step(state, input) admits a frame, finishes an invocation, applies a notification or a registry change, and returns the next state with the Effects to perform: bytes to write on an exchange, an Invocation to run, an invocation to cancel, and an exchange to close. It performs no I/O. perform(invocation, report_progress) runs the handler and returns the Finished input to feed back. relay/runtime is the supervised interpreter of these effects that the bundled transports use; prefer it unless you need to own the scheduling.

A transport allocates an ExchangeId with new_exchange_id before parsing, so that a malformed frame still has a destination. Input and Effect are revision-bound: they gain variants only in a major release. The state keeps up to 10,000 closed exchange records to suppress late or repeated output, and drops the oldest beyond that.

import gleam/option.{None}
import relay/reducer
import relay/server

pub fn admit(bytes: BitArray) -> List(reducer.Effect(Nil)) {
  let state = reducer.init(server.new([]))
  let exchange = reducer.new_exchange_id()
  let #(_state, effects) =
    reducer.step(state, reducer.Received(exchange, Nil, bytes, None))
  effects
}

Types

Effect

</>

What the interpreter must do.

pub type Effect(context) {
  Write(exchange: ExchangeId, bytes: BitArray)
  Start(invocation: Invocation(context))
  Cancel(invocation: InvocationId)
  Close(exchange: ExchangeId)
  Admitted(exchange: ExchangeId, method: String)
}

Constructors

Write(exchange: ExchangeId, bytes: BitArray)

```
Start(invocation: Invocation(context))
```
```
Cancel(invocation: InvocationId)
```
```
Close(exchange: ExchangeId)
```
```
Admitted(exchange: ExchangeId, method: String)
```
The server admitted a request on this exchange; for telemetry.

ExchangeId

opaque </>

The destination of one inbound frame and everything written for it.

pub opaque type ExchangeId

Input

</>

What the reducer reacts to.

pub type Input(context) {
  Received(
    exchange: ExchangeId,
    context: context,
    bytes: BitArray,
    correlation: option.Option(correlation.Correlation),
  )
  Finished(invocation: InvocationId, outcome: Outcome)
  Progressed(
    invocation: InvocationId,
    progress: Float,
    total: option.Option(Float),
    message: option.Option(String),
  )
  Crashed(invocation: InvocationId)
  TimedOut(invocation: InvocationId)
  ExchangeClosed(exchange: ExchangeId)
  Notify(notification: subscriptions.Notification)
  RegisterTool(tool: tool.Tool(context))
  UnregisterTool(name: String)
  EndStreams
}

Constructors

```
Received(
  exchange: ExchangeId,
  context: context,
  bytes: BitArray,
  correlation: option.Option(correlation.Correlation),
)
```
A frame arrived on a new exchange, with the context and correlation the transport built for it. With None, the request uses the correlation the frame carries in _meta, or a fresh one.

Finished(invocation: InvocationId, outcome: Outcome)

Progressed(
  invocation: InvocationId,
  progress: Float,
  total: option.Option(Float),
  message: option.Option(String),
)

```
Crashed(invocation: InvocationId)
```
The handler crashed; the client receives an internal error.
```
TimedOut(invocation: InvocationId)
```
The handler ran out of time; the client receives an internal error.
```
ExchangeClosed(exchange: ExchangeId)
```
The peer closed this exchange; its invocation is cancelled.
```
Notify(notification: subscriptions.Notification)
```
Tells every listening stream that asked for this notification.
```
RegisterTool(tool: tool.Tool(context))
```
```
UnregisterTool(name: String)
```
```
EndStreams
```
Ends every open subscriptions/listen stream with its result.

Invocation

opaque </>

Handler work admitted by the reducer, to run outside it.

pub opaque type Invocation(context)

InvocationId

opaque </>

One admitted unit of handler work.

pub opaque type InvocationId

Outcome

opaque </>

The result of perform, fed back with Finished.

pub opaque type Outcome

State

opaque </>

The reducer state of one connection.

pub opaque type State(context)

Values

exchange_id_to_int

</>

pub fn exchange_id_to_int(id: ExchangeId) -> Int

The integer telemetry reports for an exchange.

init

</>

pub fn init(server: server.Server(context)) -> State(context)

The state of a connection to this server.

invocation_context

</>

pub fn invocation_context(
  invocation: Invocation(context),
) -> context

The application context the transport built for the request.

invocation_correlation

</>

pub fn invocation_correlation(
  invocation: Invocation(context),
) -> correlation.Correlation

The request’s correlation: the transport’s, else the one the frame carried, else a fresh one.

invocation_exchange

</>

pub fn invocation_exchange(
  invocation: Invocation(context),
) -> ExchangeId

The exchange the invocation answers on.

invocation_id

</>

pub fn invocation_id(
  invocation: Invocation(context),
) -> InvocationId

The invocation’s id.

invocation_id_to_int

</>

pub fn invocation_id_to_int(id: InvocationId) -> Int

The integer telemetry reports for an invocation.

invocation_method

</>

pub fn invocation_method(
  invocation: Invocation(context),
) -> String

The JSON-RPC method, such as "tools/call".

invocation_tool

</>

pub fn invocation_tool(
  invocation: Invocation(context),
) -> option.Option(String)

The tool name of a tools/call.

new_exchange_id

</>

pub fn new_exchange_id() -> ExchangeId

A fresh exchange id, unique within the node.

open_streams

</>

pub fn open_streams(state: State(context)) -> Int

The number of open subscriptions/listen streams.

outcome_status

</>

pub fn outcome_status(outcome: Outcome) -> telemetry.Status

How the invocation behind an outcome finished.

perform

</>

pub fn perform(
  invocation: Invocation(context),
  report_progress: fn(
    Float,
    option.Option(Float),
    option.Option(String),
  ) -> Nil,
) -> Input(context)

Runs the invocation’s handler in the calling process and returns the Finished input for the reducer. report_progress receives each progress report; the caller feeds it back as Progressed. Call it from the process that signal_cancelled will address.

server

</>

pub fn server(state: State(context)) -> server.Server(context)

The server description, after any registry changes.

signal_cancelled

</>

pub fn signal_cancelled(
  worker: process.Pid,
  invocation: InvocationId,
) -> Nil

Tells the process running an invocation that it was cancelled, which fires its handler’s relay/tool.cancelled selector.

step

</>

pub fn step(
  state: State(context),
  input: Input(context),
) -> #(State(context), List(Effect(context)))

Reduces one input.