Skip to main content

Event Stream

The event stream is the primary way for bots to receive notifications about challenges, games starting, and game endings.

GET /api/stream/event

Stream incoming events in real-time. This is a long-lived HTTP connection that sends newline-delimited JSON (ndjson).

Authentication

Requires OAuth token with bot:play, board:play, or challenge:read scope.

Request

curl -N https://lichess.org/api/stream/event \
  -H "Authorization: Bearer <your_token>"

Event Types

gameStart
object
A game has started. Connect to the game stream immediately.
gameFinish
object
A game has finished.
challenge
object
Someone challenged you.
challengeDeclined
object
A challenge was declined.
challengeCanceled
object
A challenge was canceled.
{
  "type": "gameStart",
  "game": {
    "id": "5IrD6Gzz",
    "compat": {
      "bot": true,
      "board": false
    }
  }
}

Game Stream

GET /api/bot/game/stream/

Stream the state of a game being played with the Bot API. Provides real-time updates for moves, draw offers, takeback requests, and opponent presence.
Connect to this stream as soon as you receive a gameStart event.

Authentication

Requires OAuth token with bot:play scope.

Parameters

gameId
string
required
The game ID to stream

Request

curl -N https://lichess.org/api/bot/game/stream/5IrD6Gzz \
  -H "Authorization: Bearer <your_token>"

Event Types

gameFull
object
Full game data. This is always the first message sent. Contains both game metadata and current state.
gameState
object
Current state of the game. Sent after each move.
chatLine
object
A chat message.
opponentGone
object
The opponent left the game.
{
  "type": "gameFull",
  "id": "5IrD6Gzz",
  "variant": {
    "key": "standard",
    "name": "Standard",
    "short": "Std"
  },
  "speed": "blitz",
  "perf": {
    "name": "Blitz"
  },
  "rated": true,
  "createdAt": 1710331200000,
  "white": {
    "id": "player1",
    "name": "Player1",
    "rating": 1500
  },
  "black": {
    "id": "mybot",
    "name": "MyBot",
    "title": "BOT",
    "rating": 1600
  },
  "initialFen": "startpos",
  "clock": {
    "initial": 180000,
    "increment": 2000
  },
  "state": {
    "type": "gameState",
    "moves": "e2e4",
    "wtime": 180000,
    "btime": 178000,
    "winc": 2000,
    "binc": 2000,
    "status": "started"
  }
}

Make a Move

POST /api/bot/game//move/

Make a move in a game being played with the Bot API.

Authentication

Requires OAuth token with bot:play scope.

Parameters

gameId
string
required
The game ID
move
string
required
The move in UCI format (e.g., e2e4, e7e5, e1g1 for castling)
offeringDraw
boolean
Whether to offer a draw with this move

Request

curl -X POST https://lichess.org/api/bot/game/5IrD6Gzz/move/e7e5 \
  -H "Authorization: Bearer <your_token>"

# Offer a draw
curl -X POST "https://lichess.org/api/bot/game/5IrD6Gzz/move/e7e5?offeringDraw=true" \
  -H "Authorization: Bearer <your_token>"

Response

ok
boolean
Whether the move was successful
{
  "ok": true
}

UCI Move Format

Moves must be in Universal Chess Interface (UCI) format:
  • Normal moves: e2e4, g8f6
  • Castling: e1g1 (kingside), e1c1 (queenside)
  • Promotion: e7e8q (promote to queen), a7a8n (promote to knight)
  • En passant: Use the destination square (e.g., e5d6)

Additional Game Actions

Abort Game

POST /api/bot/game//abort Abort a game. Only possible in the first few moves.
curl -X POST https://lichess.org/api/bot/game/5IrD6Gzz/abort \
  -H "Authorization: Bearer <your_token>"

Resign Game

POST /api/bot/game//resign Resign a game.
curl -X POST https://lichess.org/api/bot/game/5IrD6Gzz/resign \
  -H "Authorization: Bearer <your_token>"

Write in Chat

POST /api/bot/game//chat Post a message in the game chat.
room
string
required
Chat room: player (private) or spectator (public)
text
string
required
Message text
curl -X POST https://lichess.org/api/bot/game/5IrD6Gzz/chat \
  -H "Authorization: Bearer <your_token>" \
  -H "Content-Type: application/json" \
  -d '{"room":"player","text":"Good luck!"}'

Complete Bot Example

import requests
import json
import chess
import chess.engine
from threading import Thread

class LichessBot:
    def __init__(self, token, engine_path="/usr/games/stockfish"):
        self.token = token
        self.headers = {"Authorization": f"Bearer {token}"}
        self.base_url = "https://lichess.org"
        self.engine = chess.engine.SimpleEngine.popen_uci(engine_path)
        
    def stream_events(self):
        """Stream incoming events"""
        url = f"{self.base_url}/api/stream/event"
        response = requests.get(url, headers=self.headers, stream=True)
        
        for line in response.iter_lines():
            if line:
                event = json.loads(line)
                self.handle_event(event)
    
    def handle_event(self, event):
        """Handle incoming events"""
        event_type = event.get('type')
        
        if event_type == 'gameStart':
            game_id = event['game']['id']
            print(f"Game {game_id} started")
            # Start a new thread to handle this game
            Thread(target=self.play_game, args=(game_id,)).start()
        
        elif event_type == 'challenge':
            challenge = event['challenge']
            print(f"Challenge from {challenge['challenger']['name']}")
            # Auto-accept challenges (add your own logic here)
            self.accept_challenge(challenge['id'])
    
    def play_game(self, game_id):
        """Play a single game"""
        url = f"{self.base_url}/api/bot/game/stream/{game_id}"
        response = requests.get(url, headers=self.headers, stream=True)
        
        board = chess.Board()
        our_color = None
        
        for line in response.iter_lines():
            if line:
                event = json.loads(line)
                
                if event['type'] == 'gameFull':
                    # Determine our color
                    state = event['state']
                    our_color = self.get_our_color(event)
                    self.update_board(board, state['moves'])
                    
                    if self.is_our_turn(board, our_color):
                        self.make_best_move(game_id, board)
                
                elif event['type'] == 'gameState':
                    self.update_board(board, event['moves'])
                    
                    if event['status'] != 'started':
                        print(f"Game over: {event['status']}")
                        break
                    
                    if self.is_our_turn(board, our_color):
                        self.make_best_move(game_id, board)
                
                elif event['type'] == 'chatLine':
                    print(f"Chat: {event['username']}: {event['text']}")
    
    def update_board(self, board, moves_str):
        """Update board with move string"""
        board.reset()
        if moves_str:
            for move in moves_str.split():
                board.push_uci(move)
    
    def get_our_color(self, game_full):
        """Determine which color we are playing"""
        # Check if we're white or black based on player IDs
        # (implementation depends on how you store bot ID)
        return chess.WHITE  # Simplified
    
    def is_our_turn(self, board, our_color):
        """Check if it's our turn"""
        return board.turn == our_color
    
    def make_best_move(self, game_id, board):
        """Calculate and make the best move"""
        result = self.engine.play(board, chess.engine.Limit(time=1.0))
        move_uci = result.move.uci()
        
        url = f"{self.base_url}/api/bot/game/{game_id}/move/{move_uci}"
        response = requests.post(url, headers=self.headers)
        
        if response.status_code == 200:
            print(f"Played: {move_uci}")
        else:
            print(f"Move error: {response.json()}")
    
    def accept_challenge(self, challenge_id):
        """Accept a challenge"""
        url = f"{self.base_url}/api/challenge/{challenge_id}/accept"
        requests.post(url, headers=self.headers)
    
    def run(self):
        """Start the bot"""
        print("Bot starting...")
        try:
            self.stream_events()
        finally:
            self.engine.quit()

# Run the bot
if __name__ == "__main__":
    bot = LichessBot("lip_yourtoken")
    bot.run()

Best Practices

Connection Management

  1. Keep Streams Alive: The event and game streams are long-lived connections
  2. Handle Disconnections: Implement reconnection logic with exponential backoff
  3. Process Events Quickly: Don’t block the stream processing thread
  4. Use Threads: Handle each game in a separate thread

Performance

  1. Think Time: Respect time controls and don’t take too long to move
  2. Memory: Clean up finished games to avoid memory leaks
  3. Rate Limits: Avoid making too many API calls

Fair Play

  1. Engine Strength: Don’t make your bot play at unrealistic strength
  2. No Unfair Advantage: Follow Lichess Terms of Service
  3. Handle Draws: Respond appropriately to draw offers

See Also

Build docs developers (and LLMs) love