Bluesky DM API: Messaging with AT Protocol

The Bluesky API provides a powerful foundation for building social applications, and its direct messaging capabilities open up exciting possibilities for developers. Unlike traditional social platforms with proprietary APIs, Bluesky's AT Protocol messaging system is built on open standards, giving you complete control over how you implement chat features.

In this guide, you'll learn how to integrate Bluesky direct messages into your applications using TypeScript. We'll cover everything from authentication to sending messages, handling rich text, and implementing real-time updates.

Understanding the Decentralized Architecture

Before writing any code, it's essential to understand how Bluesky's messaging system differs from centralized platforms. The AT Protocol separates concerns across multiple services:

The Bluesky DM API routes requests through your Personal Data Server (PDS), which then proxies them to the chat service. This architecture means you can't simply hit a single endpoint. Instead, you need to resolve the user's PDS first.

interface BlueskySession {
  accessJwt: string;
  refreshJwt: string;
  did: string;
  handle: string;
}

async function resolvePdsBaseUrl(userDid: string): Promise<string> {
  try {
    const plcResponse = await fetch(
      `https://plc.directory/${encodeURIComponent(userDid)}`
    );
    
    if (!plcResponse.ok) {
      throw new Error(`PLC lookup failed: ${plcResponse.status}`);
    }
    
    const plcDoc = await plcResponse.json();
    const services = plcDoc?.service || [];
    
    const pdsService = services.find((service: any) =>
      service?.type?.toLowerCase().includes('atprotopersonaldataserver') ||
      service?.id?.includes('#atproto_pds')
    );
    
    if (pdsService?.serviceEndpoint) {
      return pdsService.serviceEndpoint.replace(/\/$/, '');
    }
    
    throw new Error('No PDS endpoint found in DID document');
  } catch (error) {
    console.warn(`Failed to resolve PDS for ${userDid}:`, error);
    return 'https://bsky.social'; // Fallback to default
  }
}

Note: The PDS resolution step is critical. Skipping it will cause authentication failures when accessing the chat service, especially for users on self-hosted PDS instances.

Authentication with App Passwords

The Bluesky API uses App Passwords for authentication rather than OAuth 2.0. For DM access, you must create an App Password with explicit messaging permissions enabled.

interface AuthConfig {
  identifier: string; // Handle or email
  password: string;   // App Password (not main password)
}

async function createSession(config: AuthConfig): Promise<BlueskySession> {
  const response = await fetch(
    'https://bsky.social/xrpc/com.atproto.server.createSession',
    {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        identifier: config.identifier,
        password: config.password,
      }),
    }
  );

  if (!response.ok) {
    const errorBody = await response.text();
    throw new Error(`Authentication failed: ${response.status} ${errorBody}`);
  }

  const session = await response.json();
  
  return {
    accessJwt: session.accessJwt,
    refreshJwt: session.refreshJwt,
    did: session.did,
    handle: session.handle,
  };
}

async function refreshSession(refreshJwt: string): Promise<BlueskySession> {
  const response = await fetch(
    'https://bsky.social/xrpc/com.atproto.server.refreshSession',
    {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${refreshJwt}`,
      },
    }
  );

  if (!response.ok) {
    throw new Error('Session refresh failed. Please re-authenticate.');
  }

  return response.json();
}

When creating your App Password in Bluesky settings, make sure to check "Allow access to your direct messages." Without this permission, you'll receive XRPCNotSupported errors when calling chat endpoints.

PDS Resolution and Chat Service

The chat service requires a special proxy header to route requests correctly. Here's a complete implementation of the chat request handler:

const CHAT_SERVICE_DID = 'did:web:api.bsky.chat#bsky_chat';

interface ChatRequestOptions {
  accessToken: string;
  method: 'GET' | 'POST';
  endpoint: string;
  params?: Record<string, any>;
  body?: any;
  userDid: string;
  refreshToken?: string;
}

interface ChatResponse<T> {
  data: T;
  newTokens?: {
    accessToken: string;
    refreshToken: string;
    expiresIn: number;
  };
}

async function chatRequest<T>(
  options: ChatRequestOptions
): Promise<ChatResponse<T>> {
  const { accessToken, method, endpoint, params, body, userDid, refreshToken } = options;
  
  // Resolve user's PDS endpoint
  const pdsBaseUrl = await resolvePdsBaseUrl(userDid);
  const url = new URL(`${pdsBaseUrl}/xrpc/${endpoint}`);

  // Add query parameters for GET requests
  if (method === 'GET' && params) {
    Object.entries(params).forEach(([key, value]) => {
      if (value !== undefined && value !== null) {
        if (Array.isArray(value)) {
          value.forEach(v => url.searchParams.append(key, String(v)));
        } else {
          url.searchParams.append(key, String(value));
        }
      }
    });
  }

  const headers: Record<string, string> = {
    'Authorization': `Bearer ${accessToken}`,
    'atproto-proxy': CHAT_SERVICE_DID,
  };

  if (method === 'POST' && body) {
    headers['Content-Type'] = 'application/json';
  }

  const response = await fetch(url.toString(), {
    method,
    headers,
    body: method === 'POST' && body ? JSON.stringify(body) : undefined,
  });

  if (!response.ok) {
    const errorBody = await response.text();
    
    // Handle token expiration with auto-refresh
    if (refreshToken && errorBody.includes('ExpiredToken')) {
      const newSession = await refreshSession(refreshToken);
      
      // Retry with new token
      return chatRequest({
        ...options,
        accessToken: newSession.accessJwt,
        refreshToken: newSession.refreshJwt,
      });
    }

    // Provide helpful error messages
    if (errorBody.includes('XRPCNotSupported')) {
      throw new Error(
        'DM access not enabled. Create a new App Password with "Allow access to your direct messages" checked.'
      );
    }

    throw new Error(`Chat request failed: ${response.status} ${errorBody}`);
  }

  return { data: await response.json() };
}

Listing Conversations with the Bluesky DM API

Now let's implement the core functionality for listing conversations. The AT Protocol messaging system returns conversations with participant information and the last message:

interface Participant {
  did: string;
  handle: string;
  displayName?: string;
  avatar?: string;
}

interface Message {
  id: string;
  text: string;
  sentAt: string;
  sender: {
    did: string;
  };
}

interface Conversation {
  id: string;
  rev: string;
  unreadCount: number;
  muted: boolean;
  participants: Participant[];
  lastMessage: Message | null;
}

interface ListConversationsOptions {
  limit?: number;
  cursor?: string;
  readState?: 'unread';
  status?: 'request' | 'accepted';
}

async function listConversations(
  accessToken: string,
  userDid: string,
  options: ListConversationsOptions = {}
): Promise<{ conversations: Conversation[]; cursor?: string }> {
  const params: Record<string, any> = {
    limit: options.limit || 50,
  };
  
  if (options.cursor) params.cursor = options.cursor;
  if (options.readState) params.readState = options.readState;
  if (options.status) params.status = options.status;

  const result = await chatRequest<any>({
    accessToken,
    method: 'GET',
    endpoint: 'chat.bsky.convo.listConvos',
    params,
    userDid,
  });

  const conversations = (result.data.convos || []).map((convo: any) => ({
    id: convo.id,
    rev: convo.rev,
    unreadCount: convo.unreadCount || 0,
    muted: convo.muted || false,
    participants: (convo.members || []).map((member: any) => ({
      did: member.did,
      handle: member.handle,
      displayName: member.displayName,
      avatar: member.avatar,
    })),
    lastMessage: convo.lastMessage ? {
      id: convo.lastMessage.id,
      text: convo.lastMessage.text,
      sentAt: convo.lastMessage.sentAt,
      sender: { did: convo.lastMessage.sender?.did },
    } : null,
  }));

  return {
    conversations,
    cursor: result.data.cursor,
  };
}

// Usage example
async function displayInbox(session: BlueskySession) {
  const { conversations } = await listConversations(
    session.accessJwt,
    session.did,
    { limit: 20 }
  );

  for (const convo of conversations) {
    const otherParticipants = convo.participants
      .filter(p => p.did !== session.did)
      .map(p => p.displayName || p.handle)
      .join(', ');

    console.log(`Conversation with: ${otherParticipants}`);
    console.log(`  Unread: ${convo.unreadCount}`);
    
    if (convo.lastMessage) {
      console.log(`  Last message: ${convo.lastMessage.text.slice(0, 50)}...`);
    }
  }
}

Sending Bluesky Direct Messages

Sending messages through the Bluesky DM API requires the conversation ID. You can either use an existing conversation or create one with specific participants:

interface SendMessageResult {
  id: string;
  rev: string;
  text: string;
  sentAt: string;
}

async function sendMessage(
  accessToken: string,
  userDid: string,
  conversationId: string,
  text: string
): Promise<SendMessageResult> {
  const result = await chatRequest<any>({
    accessToken,
    method: 'POST',
    endpoint: 'chat.bsky.convo.sendMessage',
    body: {
      convoId: conversationId,
      message: { text },
    },
    userDid,
  });

  return {
    id: result.data.id,
    rev: result.data.rev,
    text: result.data.text,
    sentAt: result.data.sentAt,
  };
}

async function getOrCreateConversation(
  accessToken: string,
  userDid: string,
  memberDids: string[]
): Promise<{ id: string; rev: string }> {
  const result = await chatRequest<any>({
    accessToken,
    method: 'GET',
    endpoint: 'chat.bsky.convo.getConvoForMembers',
    params: { members: memberDids },
    userDid,
  });

  return {
    id: result.data.convo.id,
    rev: result.data.convo.rev,
  };
}

// Complete example: Send a DM to a user by handle
async function sendDirectMessage(
  session: BlueskySession,
  recipientHandle: string,
  messageText: string
): Promise<SendMessageResult> {
  // First, resolve the recipient's DID from their handle
  const resolveResponse = await fetch(
    `https://bsky.social/xrpc/com.atproto.identity.resolveHandle?handle=${encodeURIComponent(recipientHandle)}`
  );
  
  if (!resolveResponse.ok) {
    throw new Error(`Could not resolve handle: ${recipientHandle}`);
  }
  
  const { did: recipientDid } = await resolveResponse.json();

  // Get or create the conversation
  const conversation = await getOrCreateConversation(
    session.accessJwt,
    session.did,
    [recipientDid]
  );

  // Send the message
  return sendMessage(
    session.accessJwt,
    session.did,
    conversation.id,
    messageText
  );
}

Rich Text and Facets

The AT Protocol supports rich text through "facets," which are annotations that add links, mentions, and hashtags to plain text. When building messages, you need to calculate byte offsets (not character offsets) for proper rendering:

interface Facet {
  index: {
    byteStart: number;
    byteEnd: number;
  };
  features: Array<{
    $type: string;
    uri?: string;
    did?: string;
    tag?: string;
  }>;
}

function parseRichTextFacets(text: string): Facet[] {
  const facets: Facet[] = [];

  // Parse URLs
  const urlRegex = /https?:\/\/[^\s<>"{}|\\^`[\]]+/gi;
  let urlMatch;
  
  while ((urlMatch = urlRegex.exec(text)) !== null) {
    const url = urlMatch[0];
    const start = urlMatch.index;
    const end = start + url.length;

    // Convert character positions to byte positions
    const byteStart = Buffer.byteLength(text.substring(0, start), 'utf8');
    const byteEnd = Buffer.byteLength(text.substring(0, end), 'utf8');

    facets.push({
      index: { byteStart, byteEnd },
      features: [{
        $type: 'app.bsky.richtext.facet#link',
        uri: url,
      }],
    });
  }

  // Parse mentions (@handle.domain)
  const mentionRegex = /@([a-zA-Z0-9._-]+(?:\.[a-zA-Z0-9._-]+)*)/gi;
  let mentionMatch;
  
  while ((mentionMatch = mentionRegex.exec(text)) !== null) {
    const mention = mentionMatch[0];
    const handle = mentionMatch[1];
    const start = mentionMatch.index;
    const end = start + mention.length;

    const byteStart = Buffer.byteLength(text.substring(0, start), 'utf8');
    const byteEnd = Buffer.byteLength(text.substring(0, end), 'utf8');

    facets.push({
      index: { byteStart, byteEnd },
      features: [{
        $type: 'app.bsky.richtext.facet#mention',
        did: handle, // Will be resolved to actual DID
      }],
    });
  }

  // Sort by position
  facets.sort((a, b) => a.index.byteStart - b.index.byteStart);
  
  return facets;
}

// Resolve mention handles to DIDs
async function resolveMentionDids(
  accessToken: string,
  facets: Facet[]
): Promise<Facet[]> {
  const handles = new Set<string>();
  
  for (const facet of facets) {
    for (const feature of facet.features) {
      if (
        feature.$type === 'app.bsky.richtext.facet#mention' &&
        feature.did &&
        !feature.did.startsWith('did:')
      ) {
        handles.add(feature.did.toLowerCase());
      }
    }
  }

  if (handles.size === 0) return facets;

  // Batch resolve handles
  const handleArray = Array.from(handles);
  const url = new URL('https://bsky.social/xrpc/app.bsky.actor.getProfiles');
  handleArray.forEach(h => url.searchParams.append('actors', h));

  const response = await fetch(url.toString(), {
    headers: { 'Authorization': `Bearer ${accessToken}` },
  });

  const handleToDid = new Map<string, string>();
  
  if (response.ok) {
    const data = await response.json();
    for (const profile of data.profiles || []) {
      if (profile.handle && profile.did) {
        handleToDid.set(profile.handle.toLowerCase(), profile.did);
      }
    }
  }

  // Update facets with resolved DIDs
  return facets.map(facet => ({
    ...facet,
    features: facet.features
      .map(feature => {
        if (feature.$type !== 'app.bsky.richtext.facet#mention') {
          return feature;
        }
        const did = handleToDid.get(feature.did?.toLowerCase() || '');
        return did ? { ...feature, did } : null;
      })
      .filter(Boolean) as Facet['features'],
  })).filter(facet => facet.features.length > 0);
}

Note: Byte offset calculation is crucial for non-ASCII characters. A single emoji can be 4 bytes, so using character indices will break rich text rendering.

Real-time Updates

For real-time message updates, you'll need to implement polling or use the AT Protocol's event stream. Here's a practical polling implementation:

interface MessagePollerOptions {
  accessToken: string;
  userDid: string;
  conversationId: string;
  onMessage: (message: Message) => void;
  onError: (error: Error) => void;
  pollInterval?: number;
}

class MessagePoller {
  private intervalId: NodeJS.Timeout | null = null;
  private lastMessageId: string | null = null;
  private options: MessagePollerOptions;

  constructor(options: MessagePollerOptions) {
    this.options = {
      pollInterval: 5000, // 5 seconds default
      ...options,
    };
  }

  async start(): Promise<void> {
    // Get initial messages to set baseline
    const { messages } = await this.fetchMessages();
    if (messages.length > 0) {
      this.lastMessageId = messages[0].id;
    }

    this.intervalId = setInterval(async () => {
      try {
        const { messages } = await this.fetchMessages();
        
        // Find new messages
        const newMessages: Message[] = [];
        for (const msg of messages) {
          if (msg.id === this.lastMessageId) break;
          newMessages.push(msg);
        }

        if (newMessages.length > 0) {
          this.lastMessageId = newMessages[0].id;
          // Deliver in chronological order
          newMessages.reverse().forEach(msg => {
            this.options.onMessage(msg);
          });
        }
      } catch (error) {
        this.options.onError(error as Error);
      }
    }, this.options.pollInterval);
  }

  stop(): void {
    if (this.intervalId) {
      clearInterval(this.intervalId);
      this.intervalId = null;
    }
  }

  private async fetchMessages(): Promise<{ messages: Message[] }> {
    const result = await chatRequest<any>({
      accessToken: this.options.accessToken,
      method: 'GET',
      endpoint: 'chat.bsky.convo.getMessages',
      params: {
        convoId: this.options.conversationId,
        limit: 20,
      },
      userDid: this.options.userDid,
    });

    return {
      messages: (result.data.messages || [])
        .filter((msg: any) => msg.$type !== 'chat.bsky.convo.defs#deletedMessageView')
        .map((msg: any) => ({
          id: msg.id,
          text: msg.text || '',
          sentAt: msg.sentAt,
          sender: { did: msg.sender?.did },
        })),
    };
  }
}

// Usage
const poller = new MessagePoller({
  accessToken: session.accessJwt,
  userDid: session.did,
  conversationId: 'convo-123',
  onMessage: (message) => {
    console.log(`New message: ${message.text}`);
  },
  onError: (error) => {
    console.error('Polling error:', error);
  },
  pollInterval: 3000,
});

poller.start();

Working with AT URIs

The AT Protocol uses URIs in the format at://did/collection/rkey to identify records. Understanding this format is essential for working with Bluesky direct messages:

interface ATUri {
  did: string;
  collection: string;
  rkey: string;
}

function parseATUri(uri: string): ATUri {
  // Format: at://did:plc:xxx/app.bsky.feed.post/abc123
  const match = uri.match(/^at:\/\/(did:[^/]+)\/([^/]+)\/(.+)$/);
  
  if (!match) {
    throw new Error(`Invalid AT URI: ${uri}`);
  }

  return {
    did: match[1],
    collection: match[2],
    rkey: match[3],
  };
}

function buildATUri(did: string, collection: string, rkey: string): string {
  return `at://${did}/${collection}/${rkey}`;
}

// Extract rkey for deletion operations
function extractRkey(uri: string): string {
  const parts = uri.split('/');
  return parts[parts.length - 1];
}

Error Handling and Rate Limits

The Bluesky API has rate limits that vary by endpoint. Implement proper error handling and exponential backoff:

interface RateLimitInfo {
  limit: number;
  remaining: number;
  reset: Date;
}

class BlueskyApiError extends Error {
  constructor(
    message: string,
    public statusCode: number,
    public errorType: 'rate-limit' | 'auth' | 'validation' | 'server' | 'unknown',
    public rateLimitInfo?: RateLimitInfo
  ) {
    super(message);
    this.name = 'BlueskyApiError';
  }
}

function parseApiError(response: Response, body: string): BlueskyApiError {
  const lowerBody = body.toLowerCase();
  
  // Rate limiting
  if (response.status === 429 || lowerBody.includes('rate limit')) {
    const resetHeader = response.headers.get('ratelimit-reset');
    return new BlueskyApiError(
      'Rate limit exceeded. Please wait before retrying.',
      response.status,
      'rate-limit',
      resetHeader ? {
        limit: parseInt(response.headers.get('ratelimit-limit') || '0'),
        remaining: 0,
        reset: new Date(parseInt(resetHeader) * 1000),
      } : undefined
    );
  }

  // Authentication errors
  if (
    response.status === 401 ||
    lowerBody.includes('invalid token') ||
    lowerBody.includes('expired')
  ) {
    return new BlueskyApiError(
      'Authentication failed. Please reconnect your account.',
      response.status,
      'auth'
    );
  }

  // Validation errors
  if (response.status === 400 || lowerBody.includes('invalid')) {
    return new BlueskyApiError(
      `Validation error: ${body}`,
      response.status,
      'validation'
    );
  }

  // Server errors
  if (response.status >= 500) {
    return new BlueskyApiError(
      'Bluesky service temporarily unavailable.',
      response.status,
      'server'
    );
  }

  return new BlueskyApiError(body, response.status, 'unknown');
}

async function withRetry<T>(
  operation: () => Promise<T>,
  maxRetries: number = 3
): Promise<T> {
  let lastError: Error | null = null;
  
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await operation();
    } catch (error) {
      lastError = error as Error;
      
      if (error instanceof BlueskyApiError) {
        // Don't retry auth or validation errors
        if (error.errorType === 'auth' || error.errorType === 'validation') {
          throw error;
        }
        
        // Handle rate limits
        if (error.errorType === 'rate-limit' && error.rateLimitInfo) {
          const waitTime = error.rateLimitInfo.reset.getTime() - Date.now();
          if (waitTime > 0 && waitTime < 60000) {
            await new Promise(resolve => setTimeout(resolve, waitTime));
            continue;
          }
        }
      }
      
      // Exponential backoff
      const backoff = Math.min(1000 * Math.pow(2, attempt), 30000);
      await new Promise(resolve => setTimeout(resolve, backoff));
    }
  }
  
  throw lastError;
}

Using Zernio for Bluesky Integration

Building and maintaining direct integrations with the Bluesky DM API requires handling PDS resolution, token management, error handling, and keeping up with protocol changes. This complexity multiplies when you need to support multiple social platforms.

Zernio provides a unified API that abstracts away these complexities. Instead of managing separate integrations for Bluesky, Twitter, Instagram, and other platforms, you can use a single, consistent interface:

// With Zernio's unified API
import { Zernio } from '@late/sdk';

const late = new Zernio({ apiKey: process.env.LATE_API_KEY });

// Send a message across any supported platform
await late.messages.send({
  platform: 'bluesky',
  accountId: 'your-connected-account',
  conversationId: 'convo-123',
  text: 'Hello from Zernio!',
});

// List conversations with automatic token refresh
const { conversations } = await late.messages.listConversations({
  platform: 'bluesky',
  accountId: 'your-connected-account',
});

Zernio handles the infrastructure challenges for you:

• Automatic PDS Resolution: No need to query the PLC directory yourself
• Token Management: Automatic refresh and secure storage
• Rate Limit Handling: Built-in retry logic with exponential backoff
• Error Normalization: Consistent error formats across all platforms
• Webhook Support: Real-time notifications for new messages

Whether you're building a social media management tool, a customer support system, or a community platform, Zernio's unified approach lets you focus on your product instead of API maintenance.

Check out the Zernio documentation to get started with Bluesky messaging integration in minutes, not days.