Neynar Documentation

Kafka Stream has been retired

The hosted Kafka stream was shut down in August 2026 and is no longer available. For real-time Farcaster events, use webhooks — they carry the same hydrated events this stream did. For bulk or historical data, use Parquet exports or the indexer service. Reach out on Slack if you need help migrating.This page is kept for historical reference. The brokers and topics below are no longer reachable.

The Kafka stream carried the same events that power our webhook notifications. Everything below describes how it worked while it was running, and is preserved so existing integrations have a reference for what they were consuming. The topics retained 14 days of data, so the stream was never suitable for building a database of Farcaster history from day 1 — it was paired with Parquet exports for backfill. P95 data latency was under 1.5s.

Why it existed

Hub gRPC streaming delivers dehydrated events that you have to reassemble yourself to make useful (see here for example). The Kafka stream delivered a fully hydrated event (e.g., user.created) that was usable immediately. The comparison between a gRPC hub event and a Kafka event is below. Webhooks deliver these same hydrated events and are the supported way to get them today.

How it worked

The connection details, brokers and topics below are recorded as they were at shutdown. None of them accept connections any more.

Authentication used SASL/SCRAM SHA512.
The connection required TLS (sometimes called SSL for legacy reasons) for encryption.
farcaster-mainnet-events was the aggregated topic containing all events.
farcaster-mainnet-user-events carried user.created, user.updated and user.transferred
farcaster-mainnet-cast-events carried cast.created and cast.deleted
farcaster-mainnet-follow-events carried follow.created and follow.deleted
farcaster-mainnet-reaction-events carried reaction.created and reaction.deleted
farcaster-mainnet-signer-events carried signer.created and signer.deleted

Consumers could subscribe to any combination of the event-specific topics above, or to farcaster-mainnet-events to get everything in one topic. Three brokers were exposed over the Internet, and clients were given all three:

b-1-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196
b-2-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196
b-3-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196

Most clients accept the brokers as a comma-separated list:

b-2-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196,b-1-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196,b-3-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196

kcat (formerly kafkacat) was used to test connectivity locally:

brew install kcat
brew home kcat

kcat -L \
    -b 'b-2-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196,b-1-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196,b-3-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196' \
    -X security.protocol=sasl_ssl \
    -X sasl.mechanisms=SCRAM-SHA-512 \
    -X sasl.username='user-YOURNAME' \
    -X sasl.password='YOURPASSWORD'

Example output:

Metadata for farcaster-mainnet-events (from broker 1: sasl_ssl://b-1-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196/1):
 3 brokers:
  broker 2 at b-2-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196
  broker 3 at b-3-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196 (controller)
  broker 1 at b-1-public.tfmskneynar.5vlahy.c11.kafka.us-east-1.amazonaws.com:9196
 5 topics:
  topic "farcaster-mainnet-user-events" with 2 partitions:
    partition 0, leader 2, replicas: 2,3,1, isrs: 2,3,1
    partition 1, leader 3, replicas: 3,1,2, isrs: 3,1,2
  topic "farcaster-mainnet-reaction-events" with 2 partitions:
    partition 0, leader 3, replicas: 3,2,1, isrs: 3,2,1
    partition 1, leader 1, replicas: 1,3,2, isrs: 1,3,2
  topic "farcaster-mainnet-cast-events" with 2 partitions:
    partition 0, leader 2, replicas: 2,3,1, isrs: 2,3,1
    partition 1, leader 3, replicas: 3,1,2, isrs: 3,1,2
  topic "farcaster-mainnet-follow-events" with 2 partitions:
    partition 0, leader 2, replicas: 2,3,1, isrs: 2,3,1
    partition 1, leader 3, replicas: 3,1,2, isrs: 3,1,2
  topic "farcaster-mainnet-events" with 2 partitions:
    partition 0, leader 2, replicas: 2,3,1, isrs: 3,1,2
    partition 1, leader 3, replicas: 3,1,2, isrs: 3,1,2
  topic "farcaster-mainnet-signer-events" with 2 partitions:
    partition 0, leader 1, replicas: 1,3,2, isrs: 1,3,2
    partition 1, leader 2, replicas: 2,1,3, isrs: 2,1,3

The topics listed varied depending on the credentials in use.

Consumer nodejs example

https://github.com/neynarxyz/farcaster-examples/tree/main/neynar-webhook-kafka-consumer

Data schema

// _when a new user is created on the network_

interface Bio {
  text: string;
}

interface Profile {
  bio: Bio;
}

interface VerifiedAddresses {
  eth_addresses: string[];
  sol_addresses: string[];
}

interface UserCreatedData {
  object: "user";
  fid: number;
  custody_address: string;
  username: string;
  display_name: string | null;
  pfp_url: string | null;
  profile: Profile;
  follower_count: number;
  following_count: number;
  verifications: string[];
  verified_addresses: VerifiedAddresses;
  active_status: "inactive" | "active";
  power_badge: boolean;
  event_timestamp: string; // ISO 8601 format
}

interface CustomHeaders {
  "x-convoy-message-type": "broadcast";
}

interface UserCreatedEvent {
  event_type: "user.created";
  data: UserCreatedData;
  custom_headers: CustomHeaders;
  idempotency_key?: string;
}

// _when a user profile field is updated_

interface Bio {
  text: string;
}

interface Profile {
  bio: Bio;
}

interface VerifiedAddresses {
  eth_addresses: string[];
  sol_addresses: string[];
}

interface UserUpdatedData {
  object: "user";
  fid: number;
  custody_address: string;
  username: string;
  display_name: string;
  pfp_url: string;
  profile: Profile;
  follower_count: number;
  following_count: number;
  verifications: string[];
  verified_addresses: VerifiedAddresses;
  active_status: "inactive" | "active";
  power_badge: boolean;
  event_timestamp: string; // ISO 8601 format
}

interface CustomHeaders {
  "x-convoy-message-type": "broadcast";
}

interface UserUpdatedEvent {
  event_type: "user.updated";
  data: UserUpdatedData;
  custom_headers: CustomHeaders;
  idempotency_key?: string;
}

// _when a new cast is created_

export interface CustomHeaders {
  "x-convoy-message-type": "broadcast"
}

interface Fid {
  fid?: number | null;
}

interface User {
  object: string;
  fid: number;
  custody_address: string;
  username: string;
  display_name: string;
  pfp_url: string;
  profile: {
    bio: {
      text: string;
    };
  };
  follower_count: number;
  following_count: number;
  verifications: string[];
  verified_addresses: {
    eth_addresses: string[];
    sol_addresses: string[];
  };
  active_status: string;
  power_badge: boolean;
}

interface EmbedUrlMetadata {
  content_type?: string | null;
  content_length?: number | null;
}

interface EmbedUrl {
  url: string;
  metadata?: EmbedUrlMetadata;
}

interface CastId {
  fid: number;
  hash: string;
}

interface EmbedCastId {
  cast_id: CastId;
}

type EmbeddedCast = EmbedUrl | EmbedCastId;

interface EventData {
	object: "cast";
  hash: string;
  parent_hash?: string | null;
  parent_url?: string | null;
  root_parent_url?: string | null;
  parent_author?: Fid;
  author: User;
  mentioned_profiles?: User[];
  text: string;
  timestamp: string; // ISO 8601 format
  embeds: EmbeddedCast[];
}

export interface CastCreatedEvent {
  event_type: "cast.created"
  data: EventData
  custom_headers: CustomHeaders
  idempotency_key?: string
}

// _when a cast is deleted_

export interface CustomHeaders {
  "x-convoy-message-type": "broadcast"
}

interface Fid {
  fid?: number | null;
}

interface User {
  object: string;
  fid: number;
  custody_address: string;
  username: string;
  display_name: string;
  pfp_url: string;
  profile: {
    bio: {
      text: string;
    };
  };
  follower_count: number;
  following_count: number;
  verifications: string[];
  verified_addresses: {
    eth_addresses: string[];
    sol_addresses: string[];
  };
  active_status: string;
  power_badge: boolean;
}

interface EmbedUrlMetadata {
  content_type?: string | null;
  content_length?: number | null;
}

interface EmbedUrl {
  url: string;
  metadata?: EmbedUrlMetadata;
}

interface CastId {
  fid: number;
  hash: string;
}

interface EmbedCastId {
  cast_id: CastId;
}

type EmbeddedCast = EmbedUrl | EmbedCastId;

interface EventData {
	object: "cast";
  hash: string;
  parent_hash?: string | null;
  parent_url?: string | null;
  root_parent_url?: string | null;
  parent_author?: Fid;
  author: User;
  mentioned_profiles?: User[];
  text: string;
  timestamp: string; // ISO 8601 format
  embeds: EmbeddedCast[];
}

export interface CastDeletedEvent {
  event_type: "cast.deleted"
  data: EventData
  custom_headers: CustomHeaders
  idempotency_key?: string
}

// _when a user follows another user_

interface UserDehydrated {
  object: "user_dehydrated";
  fid: number;
  username: string;
}

interface AppDehydrated {
  object: "user_dehydrated";
  fid: number;
}

interface EventData {
  object: "follow";
  event_timestamp: string; // ISO 8601 format
  timestamp: string;       // ISO 8601 format with timezone
  user: UserDehydrated;
  target_user: UserDehydrated;
  app: AppDehydrated | null; // null if not signer isn't available
}

interface CustomHeaders {
  "x-convoy-message-type": "broadcast";
}

interface FollowCreatedEvent {
  event_type: "follow.created";
  data: EventData;
  custom_headers: CustomHeaders;
  idempotency_key?: string
}

// _when a user unfollows another user_

interface UserDehydrated {
  object: "user_dehydrated";
  fid: number;
  username: string;
}

interface EventData {
  object: "unfollow";
  event_timestamp: string; // ISO 8601 format
  timestamp: string;       // ISO 8601 format with timezone
  user: UserDehydrated;
  target_user: UserDehydrated;
}

interface CustomHeaders {
  "x-convoy-message-type": "broadcast";
}

interface FollowDeletedEvent {
  event_type: "follow.deleted";
  data: EventData;
  custom_headers: CustomHeaders;
  idempotency_key?: string
}

// _when a reaction is added to a cast_

interface UserDehydrated {
  object: "user_dehydrated";
  fid: number;
  username: string;
}

interface AppDehydrated {
  object: "user_dehydrated";
  fid: number;
}

interface URIDehydrated {
  object: "uri_dehydrated";
  uri: string;
}

interface CastDehydrated {
  object: "cast_dehydrated";
  hash: string;
  author: UserDehydrated;
}

interface EventData {
  object: "reaction";
  event_timestamp: string; // ISO 8601 format
  timestamp: string;       // ISO 8601 format with timezone
  reaction_type: number;
  user: UserDehydrated;
  target: CastDehydrated | URIDehydrated;
  app: AppDehydrated | null; // null if not signer isn't available
}

interface CustomHeaders {
  "x-convoy-message-type": "broadcast";
}

interface ReactionCreatedEvent {
  event_type: "reaction.created";
  data: EventData;
  custom_headers: CustomHeaders;
  idempotency_key?: string;
}

// _when a reaction is removed from a cast_

interface UserDehydrated {
  object: "user_dehydrated";
  fid: number;
  username: string;
}

interface URIDehydrated {
  object: "uri_dehydrated";
  uri: string;
}

interface CastDehydrated {
  object: "cast_dehydrated";
  hash: string;
  author: UserDehydrated;
}

interface EventData {
  object: "reaction";
  event_timestamp: string; // ISO 8601 format
  timestamp: string;       // ISO 8601 format with timezone
  reaction_type: number;
  user: UserDehydrated;
  target: CastDehydrated | URIDehydrated;
}

interface CustomHeaders {
  "x-convoy-message-type": "broadcast";
}

interface ReactionDeletedEvent {
  event_type: "reaction.deleted";
  data: EventData;
  custom_headers: CustomHeaders;
  idempotency_key?: string;
}

Get Started

Build Bots and Agents

Build Farcaster Mini Apps

Fetch User Information

Fetch Cast Information

Fetch Farcaster Feeds

Onboard New Users

Write Data to Farcaster

Filter Spam, Low Quality Data

Host Farcaster Mini Apps

Run Queries on FC Data

Render Farcaster in React

Get Events Via Webhooks

Fetch Notifications

Create Onchain Transactions

Intersect Onchain & Social Data

Write Direct Casts

Get Farcaster Storage Data

Publish Actions on FC Apps

Get Farcaster Data on Base

Explore Network Health

Get Hypersub Subscriptions

Contribute To Development

Kafka Stream: Real-Time Farcaster Events via Neynar

Kafka Stream has been retired

Why it existed

How it worked

Consumer nodejs example

https://github.com/neynarxyz/farcaster-examples/tree/main/neynar-webhook-kafka-consumer

Data schema

​Kafka Stream has been retired

​Why it existed

​How it worked

​Consumer nodejs example

https://github.com/neynarxyz/farcaster-examples/tree/main/neynar-webhook-kafka-consumer

​Data schema

Kafka Stream has been retired

Why it existed

How it worked

Consumer nodejs example

Data schema