Context Classes

PyStrands uses context objects to pass information between the Go broker and your Python backend.

Overview

There are two main context classes:

• Context — Contains client identity and room information
• ConnectionRequestContext — Contains connection request details for authentication

Both classes inherit from JSONModel which provides JSON serialization/deserialization.

Context

The Context class represents a client's session and is passed to most callback methods.

Attributes

Attribute	Type	Description
client_id	str	Unique identifier for the client (UUID v4)
room_id	str	The room the client belongs to
metadata	dict	Custom key-value storage for application data

client_id

client_id: str

A unique identifier automatically generated for each WebSocket client. This ID is used for: - Identifying clients in logs - Sending private messages - Tracking user sessions

Example:

def on_message(self, message, context):
    print(f"Message from {context.client_id}")
    # Send a private reply
    self.send_private_message(context.client_id, "Received!")

---

room_id

room_id: str

The room identifier for the client. This is typically set in on_connection_request based on the URL path.

Example:

def on_connection_request(self, request):
    # Assign room based on URL
    request.context.room_id = request.url.strip("/") or "lobby"
    return True

def on_message(self, message, context):
    # Send to everyone in the same room
    self.send_room_message(context.room_id, message)

Room Assignment

Always set room_id in on_connection_request. If not set, the default room is used.

---

metadata

metadata: dict

A dictionary for storing custom data about the client. This is useful for: - Storing authentication info (user ID, role, permissions) - Caching user preferences - Tracking connection timestamps - Any application-specific data

Example:

def on_connection_request(self, request):
    # Authenticate and store user info
    token = request.headers.get("Authorization", [""])[0]
    user = validate_token(token)

    request.context.metadata = {
        "user_id": user.id,
        "username": user.name,
        "role": user.role,
        "joined_at": time.time()
    }
    return True

def on_message(self, message, context):
    # Access metadata in handlers
    username = context.metadata.get("username", "anonymous")
    print(f"{username}: {message}")

Methods

from_json()

Class method to create a Context from a dictionary.

@classmethod
def from_json(cls, data: dict) -> Context

Parameters:

Parameter	Type	Description
data	dict	Dictionary with client_id, room_id, and metadata

Returns: Context — New Context instance

Example:

data = {
    "client_id": "550e8400-e29b-41d4-a716-446655440000",
    "room_id": "room1",
    "metadata": {"user": "alice"}
}
context = Context.from_json(data)

---

to_json()

Serialize the Context to a dictionary.

def to_json(self) -> dict

Returns: dict — Dictionary representation of the context

Example:

def on_new_connection(self, context):
    # Log context as JSON
    logger.info(f"New connection: {context.to_json()}")

ConnectionRequestContext

The ConnectionRequestContext class is passed to on_connection_request and contains all information about an incoming WebSocket connection request.

Attributes

Attribute	Type	Description
headers	Dict[str, List[str]]	HTTP headers from the WebSocket upgrade request
url	str	URL path the client connected to
remote_addr	str	Client's IP address
context	Context	The mutable context object for the connection
accepted	bool	Whether to accept the connection (default: True)

headers

headers: Dict[str, List[str]]

HTTP headers from the WebSocket upgrade request. Headers are stored as a dictionary where values are lists (since HTTP allows multiple values for the same header).

Common headers to check: - Authorization — For tokens/JWT - Cookie — For session cookies - User-Agent — For client identification - X-* — Custom application headers

Example:

def on_connection_request(self, request):
    # Get Authorization header
    auth_list = request.headers.get("Authorization", [])
    auth_header = auth_list[0] if auth_list else None

    # Get custom header
    custom_list = request.headers.get("X-API-Key", [])
    api_key = custom_list[0] if custom_list else None

    # Validate
    if not validate_auth(auth_header):
        return False

    return True

---

url

url: str

The URL path the client connected to. This is typically used to determine which room to assign the client to.

Example:

# Client connects to: ws://broker/room1
def on_connection_request(self, request):
    # request.url == "/room1"
    room = request.url.strip("/")
    request.context.room_id = room or "default"
    return True

---

remote_addr

remote_addr: str

The client's IP address. Useful for: - IP-based rate limiting - Geographic restrictions - Audit logging - Debugging

Example:

def on_connection_request(self, request):
    client_ip = request.remote_addr

    # Check blocklist
    if client_ip in BLOCKED_IPS:
        logger.warning(f"Blocked connection from {client_ip}")
        return False

    # Rate limiting
    if not check_rate_limit(client_ip):
        return False

    return True

---

context

context: Context

The Context object for this connection. Modify this to: - Set the room_id - Add metadata - Access the auto-generated client_id

Example:

def on_connection_request(self, request):
    # Access the context
    ctx = request.context

    # Set room
    ctx.room_id = request.url.strip("/")

    # Add metadata
    ctx.metadata = {
        "connected_at": time.time(),
        "ip": request.remote_addr
    }

    # Access client_id (auto-generated)
    print(f"New client: {ctx.client_id}")

    return True

---

accepted

accepted: bool

Whether to accept the connection. Defaults to True. You can: - Return True/False from on_connection_request - Or set request.accepted = False directly

Example:

# Option 1: Return boolean
def on_connection_request(self, request):
    if not is_valid(request):
        return False
    return True

# Option 2: Modify accepted directly
def on_connection_request(self, request):
    if not is_valid(request):
        request.accepted = False
    # Can still do more processing...
    return request.accepted

Methods

from_json()

Class method to create a ConnectionRequestContext from a dictionary.

@classmethod
def from_json(cls, data: dict) -> ConnectionRequestContext

Parameters:

Parameter	Type	Description
data	dict	Dictionary with headers, url, remote_addr, context, accepted

Returns: ConnectionRequestContext — New ConnectionRequestContext instance

---

to_json()

Serialize the ConnectionRequestContext to a dictionary.

def to_json(self) -> dict

Returns: dict — Dictionary representation

JSONModel Base Class

Both Context and ConnectionRequestContext inherit from JSONModel which provides:

from_json()

@classmethod
def from_json(cls, data: dict) -> T

Creates an instance from a dictionary, mapping keys to annotated attributes.

to_json()

def to_json(self) -> dict

Serializes the object to a dictionary, including all annotated attributes.

Usage Examples

Complete Authentication Example

def on_connection_request(self, request):
    # Extract info from request
    token = request.headers.get("Authorization", [""])[0]
    room = request.url.strip("/") or "lobby"
    ip = request.remote_addr

    # Validate token
    user = self.validate_token(token)
    if not user:
        logger.warning(f"Invalid token from {ip}")
        return False

    # Set up context
    request.context.room_id = room
    request.context.metadata = {
        "user_id": user.id,
        "username": user.name,
        "role": user.role,
        "ip": ip,
        "connected_at": time.time()
    }

    logger.info(f"User {user.name} joined room {room}")
    return True

def on_message(self, message, context):
    # Access metadata
    username = context.metadata["username"]
    role = context.metadata["role"]

    # Check permissions
    if role != "admin" and message.startswith("/"):
        self.send_private_message(context.client_id, "Admin only!")
        return

    # Process message
    self.send_room_message(context.room_id, f"{username}: {message}")

def on_disconnect(self, context):
    # Clean up
    username = context.metadata.get("username", "unknown")
    room = context.room_id
    logger.info(f"User {username} left room {room}")

Debugging Context

def on_connection_request(self, request):
    # Log all request info for debugging
    print(f"Connection request:")
    print(f"  URL: {request.url}")
    print(f"  IP: {request.remote_addr}")
    print(f"  Headers: {request.headers}")
    print(f"  Client ID: {request.context.client_id}")

    return True

def on_message(self, message, context):
    # Log full context
    print(f"Message context: {context.to_json()}")
    # Output: {"client_id": "...", "room_id": "...", "metadata": {...}}

Type Hints

For type checking and IDE support:

from pystrands import AsyncPyStrandsClient
from pystrands.context import Context, ConnectionRequestContext

class TypedBackend(AsyncPyStrandsClient):
    async def on_connection_request(self, request: ConnectionRequestContext) -> bool:
        # IDE knows request has headers, url, remote_addr, context, accepted
        return True

    async def on_message(self, message: str, context: Context) -> None:
        # IDE knows context has client_id, room_id, metadata
        pass

See Also

• PyStrandsClient — Synchronous client reference
• AsyncPyStrandsClient — Asynchronous client reference
• Authentication Guide — Detailed authentication patterns