Kite

Kite is an atproto Jetstream V2 subscriber library. Use it to subscribe to a jetstream server and get a live stream of records being created, updated, and deleted on the AT protocol that powers services like Bluesky, Tangled, and Leaflet.

You can either connect without a cursor to just get live data as it comes through, or define callbacks for persisted cursors to keep track of your position and never miss an event. Jetstream also supports defining filters on collections, such as app.bsky.feed.post for Bluesky posts or site.standard.document for standard.site blog posts, and even filtering by specific account identifiers, aka dids. If your handler raises, the subscriber will restart from the last known cursor, which may include records you have already processed. In other words, event delivery is at least once.

Kite implements websocket keep alive with ping/pong heartbeats and gracefully restarts on closed connections, with backoff. If you're falling behind on processing events, Kite will apply back pressure, but if you fall too far behind the upstream will reject the connection and Kite automatically falls back to requesting the oldest data available.

Kite requires Elixir 1.18 (released 2024) and OTP 25 or higher because it uses the built-in JSON module, and other modern niceties like Process.set_label (1.17). When zstd compression lands that optional feature will require OTP 28.

Kite is used live in production at shelf.cafe.

Define your compile time options and implement the available callbacks to get a Jetstream V2 compatible subscriber.

Example

defmodule MyApp.Posts do
  use Kite,
    endpoint: "wss://jetstream.us-east.bsky.network",
    collections: ["app.bsky.feed.post"],
    kinds: [:commit]

  def handle_event(payload, _context), do: IO.inspect(payload)
end

Add your newly defined subscriber to your application.ex and you're ready to go.

@impl true
def start(_type, _args) do

  children = [
    {MyApp.Posts, []} # <- Add your new module to the list of children.
    ...
  ]

  opts = [strategy: :one_for_one, name: MyApp.Supervisor]
  Supervisor.start_link(children, opts)
end

Or, for some more ad-hoc experimenting, try MyApp.Posts.start_link([]) in IEx.

Callbacks

handle_event/2 is required. If you don't define get_cursor/0 and put_cursor/1 the subscriber always starts from the latest event available.

For most practical use cases you want to implement get_cursor/0 and put_cursor/1, in order to gracefully handle restarts or intermittent issues.

The second argument of handle_event is context, which is user defined data passed when starting the subscriber. This can be used to make metadata available to the handler where you run multiple instances of the same implementation.

Options

The supported options are:

• endpoint - required - a websocket URL for a jetstream, like "wss://jetstream.us-east.bsky.network" or "wss://jetstream2.fr.hose.cam/".
• collections - a list of NSID strings representing the resources you're interested in, like "app.bsky.feed.post" or "site.standard.document".
• kinds - a list of atoms corresponding to the types of events you want to see, normally just [:commit].
• context - use this to pass anything you want to the handle_event callback, it will be passed as the second argument.
• enabled - default true, set to false to prevent the subscriber from connecting to the jetstream, for example in tests.
• flush_interval - default 5s, the frequency with which to write the latest cursor state with put_cursor/1. Write more frequently to reduce the window of potential repeated events on a crash.

Anything set in the use Kite options can be overriden when starting the subscriber from your Application children list, eg:

children = [
  {MyApp.Posts, endpoint: System.fetch_env!("JETSTREAM_URL")}
]

Payloads

Jetstream outputs events in JSON, sending batches over the websocket, and Kite exposes the raw data for you. Jetstream emits 4 different "kinds" of events: #identity, #account, #commit, and #sync. For a full reference, check out the docs. For now we'll focus on #commit which represents changes to atproto records, including create, update, and delete. Note that records can have defined schemas, aka lexicons, but the data in the jetstream is not guaranteed to be validated and correct, so you must do your own validation.

Here's an example Bluesky post (#commit kind), in the full handler payload format:

%{
  "$type" => "network.bsky.jetstream.subscribeEvents#commit",
  "cid" => "bafyreigkx5h...",
  "collection" => "app.bsky.feed.post",
  "did" => "did:plc:user",
  "operation" => "create",
  "record" => %{
    "$type" => "app.bsky.feed.post",
    "createdAt" => "2026-09-16T18:13:33.933Z",
    "langs" => ["nl"],
    "reply" => %{
      "parent" => %{
        "cid" => "bafyreigkx5h...",
        "uri" => "at://did:plc:user/app.bsky.feed.post/rkey"
      },
      "root" => %{
        "cid" => "bafyreigkx5h...",
        "uri" => "at://did:plc:user/app.bsky.feed.post/rkey"
      }
    },
    "text" => "Voeg hier even vvd aan toe."
  },
  "rev" => "3mvn...",
  "rkey" => "3mvn...",
  "seq" => 25947795773,
  "time" => "2026-09-16T18:13:35.185499Z"
}

And to demonstrate how to write a handler for it, let's say we want to just print the text of the posts.

def handle_event(%{"operation" => "create"} = commit, _context) do
  %{"record" => record, "did" => did} = commit

  if text = record["text"] do
    IO.inspect(text)
  end
end

def handle_event(_commit, _context), do: :ok

Todo

• [ ] Optional zstd compression using OTP 28's :zstd
• [ ] Batch processing
• [ ] Optional structs for the event metadata for commit, account, etc