Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.cocobase.buzz/llms.txt

Use this file to discover all available pages before exploring further.

​
Event Handlers

Your WebSocket function must define these event handlers:

​
on_connect()

Called when a player connects to the WebSocket. 
async def on_connect():
    # Access connection data
    player_name = request.get('player_name', 'Guest')
    room_id = request.get('room_id', 'default')

    # Join a room
    room.join(room_id, max_players=10)

    # Initialize state
    if not room.state:
        room.state = {'players': {}}

    # Add player
    room.state['players'][session.player_id] = {
        'name': player_name
    }

    # Broadcast to others
    await room.broadcast({
        'type': 'player_joined',
        'player_id': session.player_id
    })

    # Return welcome message (sent only to this player)
    return {
        'type': 'welcome',
        'your_id': session.player_id,
        'players': room.state['players']
    }
Returns: Dictionary that will be sent to the connecting player

​
on_message()

Called when a player sends a message to the server. 
async def on_message():
    # Get message data
    action = request.get('action')
    data = request.get('data')

    # Process based on action
    if action == 'move':
        x = request.get('x')
        y = request.get('y')

        # Update state
        room.state['players'][session.player_id]['x'] = x
        room.state['players'][session.player_id]['y'] = y

        # Broadcast to all except sender
        await room.broadcast({
            'type': 'player_moved',
            'player_id': session.player_id,
            'x': x,
            'y': y
        }, exclude=[session.player_id])

    # Return response (sent only to sender)
    return {'status': 'ok'}
Returns: Dictionary that will be sent back to the message sender

​
on_disconnect()

Called when a player disconnects (closes WebSocket or loses connection). 
async def on_disconnect():
    # Clean up player data
    if session.player_id in room.state['players']:
        player_name = room.state['players'][session.player_id]['name']
        del room.state['players'][session.player_id]

        # Notify others
        await room.broadcast({
            'type': 'player_left',
            'player_id': session.player_id,
            'player_name': player_name
        })

    # Destroy room if empty
    if room.get_player_count() == 0:
        room.destroy()

​
on_tick()

Called repeatedly at a fixed rate (default 20 times per second) for continuous game updates. 
async def on_tick():
    # Update game state
    current_time = time.time()
    delta_time = current_time - room.state.get('last_update', current_time)
    room.state['last_update'] = current_time

    # Example: Move objects
    for obj in room.state.get('objects', []):
        obj['x'] += obj['vx'] * delta_time
        obj['y'] += obj['vy'] * delta_time

    # Broadcast updates
    await room.broadcast({
        'type': 'state_update',
        'objects': room.state['objects']
    })
To use on_tick, you must start the game loop by calling room.start_game_loop() in your on_connect handler.

​
Available Objects

​
session

Access player session data and information.

​
Properties

​
player_id
string
Unique identifier for this player
​
user
object | null
Authenticated user object (if JWT token provided)
​
room_id
string
Current room ID the player is in
​
connected_at
datetime
Timestamp when player connected
​
metadata
dict
Custom session metadata dictionary

​
Methods

​
set(key, value)
function
Store custom session data
session.set('score', 100)
session.set('inventory', ['sword', 'shield'])
​
get(key, default=None)
function
Retrieve custom session data
score = session.get('score', 0)
inventory = session.get('inventory', [])

​
room

Manage rooms and broadcast messages to players.

​
Properties

​
id
string
Current room ID
​
state
dict
Shared room state accessible to all players. Initialize this in on_connect.
# Initialize
if not room.state:
    room.state = {'players': {}, 'game_phase': 'waiting'}

# Access/modify
room.state['game_phase'] = 'playing'
room.state['players'][session.player_id] = {'score': 0}
​
config
dict
Room configuration (tick_rate, custom flags, etc.)
room.config['tick_rate'] = 30  # 30 updates per second
room.config['game_started'] = True
​
metadata
dict
Custom room metadata for room listing
room.metadata['game_mode'] = 'battle_royale'
room.metadata['map'] = 'desert'
room.metadata['public'] = True

​
Methods

​
join(room_id, max_players=None)
function
Join a room (auto-created if doesn’t exist)
room.join('lobby', max_players=10)
room.join('private-room-123', max_players=2)
​
broadcast(message, exclude=None, include=None)
async function
Send message to all players in room
# Broadcast to everyone
await room.broadcast({'type': 'game_started'})

# Exclude specific players
await room.broadcast(
    {'type': 'player_moved', 'x': 100, 'y': 200},
    exclude=[session.player_id]
)

# Send only to specific players
await room.broadcast(
    {'type': 'secret_message'},
    include=['player-id-1', 'player-id-2']
)
​
send_to(player_id, message)
async function
Send message to a specific player
await room.send_to('player-id-123', {
    'type': 'private_message',
    'text': 'Hello!'
})
​
get_players()
function
Get list of connected player IDs
player_ids = room.get_players()
# ['player-1', 'player-2', 'player-3']
​
get_player_count()
function
Get number of connected players
count = room.get_player_count()  # 3
​
start_game_loop()
function
Start the game loop (enables on_tick calls)
# In on_connect
if not room.config.get('game_loop_started'):
    room.config['game_loop_started'] = True
    room.config['tick_rate'] = 20  # 20 ticks/second
    room.start_game_loop()
​
stop_game_loop()
function
Stop the game loop
room.stop_game_loop()
​
destroy()
function
Destroy the room and disconnect all players
if room.get_player_count() == 0:
    room.destroy()

​
request

Access incoming message data from the client.

​
Methods

​
get(key, default=None)
function
Get value from incoming message
action = request.get('action')
x = request.get('x', 0)
player_name = request.get('player_name', 'Guest')
​
json()
function
Get entire message as dictionary
full_message = request.json()
# {'action': 'move', 'x': 100, 'y': 200}

​
db

Access your project’s database (same as HTTP functions). 
# Create document
user = await db.collection('users').insert({
    'name': 'Alice',
    'score': 100
})

# Query documents
high_scores = await db.collection('users').where('score', '>', 50).get()

# Update document
await db.collection('users').update(user_id, {'score': 150})

# Delete document
await db.collection('users').delete(user_id)
See the Database Documentation for full database API reference.

​
Built-in Modules

The following Python modules are available in your functions:
​
time
module
Time-related functions
current_time = time.time()
time.sleep(1)
​
random
module
Random number generation
rand_num = random.random()
rand_int = random.randint(1, 10)
choice = random.choice(['a', 'b', 'c'])
​
json
module
JSON encoding/decoding
data = json.loads('{"key": "value"}')
string = json.dumps({'key': 'value'})
​
uuid
module
UUID generation
unique_id = str(uuid.uuid4())
​
datetime
module
Date and time handling
now = datetime.now()
timestamp = datetime.fromtimestamp(1234567890)

​
Rate Limiting

Each player is limited to 60 messages per second to prevent abuse. Messages exceeding this limit will receive an error response: 
{
  "error": "Rate limit exceeded",
  "message": "Maximum 60 messages per second"
}

​
Room Listing

List all active rooms for your project: 
curl http://localhost:8000/ws/rooms/YOUR_PROJECT_ID
Response: 
{
  "rooms": [
    {
      "room_id": "lobby",
      "player_count": 5,
      "max_players": 10,
      "created_at": "2024-01-15T10:30:00Z",
      "game_mode": "battle_royale",
      "map": "desert",
      "status": "waiting",
      "custom_data": {}
    }
  ],
  "total": 1
}