Snodo.Subscription (snodo v0.4.0)

An opened, request-scoped MCP subscription.

Applications implement Snodo.Subscription.Source; this module owns filter narrowing, lifecycle safety, bounded pulling, and protocol wire shaping.

A subscription outlives its request, so it keeps only what it serves. The source's open/3 callback receives the full request context; the context stored in the subscription has empty request_params and input_responses and no request_state, and accepted_filter holds the notifications it delivers. Filter strings are copied out of the request body, so an open stream does not keep the body alive.

Summary

Types

id()

t()

Functions

acknowledgement(subscription)

Returns the first required acknowledgement notification.

close(subscription, reason)

Closes the application-owned source handle with an explicit reason.

completion(subscription)

Returns the final successful JSON-RPC response for graceful completion.

failure(subscription, reason)

Returns a terminal JSON-RPC error response for a failed open stream.

notification(subscription, event)

Shapes one requested source event or drops an event outside the accepted filter.

open(source, requested_filter, context)

Opens source for the notifications in requested_filter that the server supports, and returns the opened subscription.

start_worker(subscription, owner)

Starts a monitored, demand-driven source worker owned by owner.

stop_worker(worker, monitor)

Stops a worker and removes its process monitor.

Types

id()

@type id() :: integer() | String.t()

t()

@type t() :: %Snodo.Subscription{
  accepted_filter: map(),
  context: Snodo.Context.t(),
  extension_filters: %{optional(String.t()) => map()},
  extension_registry: Snodo.Extension.Registry.t(),
  handle: term(),
  id: id(),
  protocol: module(),
  source: Snodo.Subscription.Source.Config.t()
}

Functions

acknowledgement(subscription)

@spec acknowledgement(t()) :: {:ok, map()} | {:error, Snodo.Error.t()}

Returns the first required acknowledgement notification.

close(subscription, reason)

@spec close(t(), Snodo.Subscription.Source.close_reason()) :: :ok

Closes the application-owned source handle with an explicit reason.

completion(subscription)

@spec completion(t()) :: {:ok, map()} | {:error, Snodo.Error.t()}

Returns the final successful JSON-RPC response for graceful completion.

failure(subscription, reason)

@spec failure(t(), term()) :: map()

Returns a terminal JSON-RPC error response for a failed open stream.

notification(subscription, event)

@spec notification(t(), Snodo.Subscription.Event.t()) ::
  {:ok, map()} | :drop | {:error, Snodo.Error.t()}

Shapes one requested source event or drops an event outside the accepted filter.

open(source, requested_filter, context)

@spec open(Snodo.Subscription.Source.Config.t(), map(), Snodo.Context.t()) ::
  {:ok, t()} | {:error, Snodo.Error.t()}

Opens source for the notifications in requested_filter that the server supports, and returns the opened subscription.

Serve it with start_worker/2.

start_worker(subscription, owner)

@spec start_worker(t(), pid()) :: {pid(), reference()}

Starts a monitored, demand-driven source worker owned by owner.

After each continue/1, the owner receives {:mcp_subscription, worker, outcome} where the outcome is {:ok, event}, :closed, or {:error, reason}.

The worker monitors owner. If owner exits for any reason before it calls stop_worker/2, the worker closes the source handle with {:disconnected, {:owner_down, reason}} and exits, so a killed transport process does not leave the source open.

stop_worker(worker, monitor)

@spec stop_worker(pid(), reference()) :: :ok

Stops a worker and removes its process monitor.

The worker exits without closing the source handle; the owner closes it with close/2.