- Snek Community
+ Snek
+ Docs
+ Architecture
+ API
+ Bots
+ Contribute
+ About
+ Learn how to use, extend, and contribute to the Snek platform
+Learn why specific technologies were chosen and the philosophy behind the platform's design.
+ + + + → +Complete WebSocket RPC API documentation with all available methods for channels, messages, and more.
+ + + + → +Build automated bots with complete Python and JavaScript examples. Integrate with AI services.
+ + + + → +Set up development environment, understand project structure, and learn how to add features.
+ + + + → +Learn about the platform's mission, principles, and why it exists.
+ + + +Get Snek running locally in under a minute:
+pip install git+https://retoor.molodetz.nl/retoor/snek.git +snek serve+
The server starts at http://localhost:8080. Register a user and start chatting.
Snek is a privacy-focused platform for developers:
+Access the platform via multiple protocols:
+davs://molodetz.online/webdavsftp://molodetz.online:2242git@molodetz.online:username/repo.gitwss://snek.community/rpc.wsAll protocols use your Snek credentials.
+Professional Platform for Developers, Testers & AI Professionals
Login Register + Docs AboutWebSocket RPC interface documentation
+Snek uses a JSON-RPC style protocol over WebSocket. Connect to /rpc.ws to establish a connection.
{
+ "callId": "unique-id",
+ "method": "method_name",
+ "args": [arg1, arg2, ...]
+}
+
+ {
+ "callId": "unique-id",
+ "success": true,
+ "data": { ... }
+}
+
+ The server pushes events without a callId:
+{
+ "channel_uid": "...",
+ "event": "new_message",
+ "data": { ... }
+}
+ Authenticate and establish a session. On success, the WebSocket connection is associated with the user and subscribed to all their channels.
+username (string) - The username
+password (string) - The password
+Returns: User record (without password field) on success, error on failure
+Keep the connection alive and update the user's last activity timestamp.
+Returns: {"pong": []}
Get user information. If no user_uid is provided, returns the current user's data.
+user_uid (string, optional) - Target user's UID
+Returns: User record (email hidden for other users)
+Search for users by username.
+query (string) - Search query
+Returns: Array of matching usernames
+Get all channels the current user is a member of.
+Returns: Array of channel objects with name, uid, tag, new_count, is_moderator, is_read_only, color
+Create a new channel. The creating user becomes the moderator.
+name (string) - Channel name
+description (string, optional) - Channel description
+is_private (boolean, optional) - Whether the channel is private
+Returns: {success: true, uid: "...", name: "..."}
Update channel settings. Only the channel creator or admin can update.
+channel_uid (string) - Channel UID
+name (string) - New channel name
+description (string, optional) - New description
+is_private (boolean, optional) - Privacy setting
+Returns: {success: true, name: "..."}
Delete a channel (soft delete). Only the channel creator or admin can delete. Public channel cannot be deleted.
+channel_uid (string) - Channel UID
+Returns: {success: true}
Invite a user to a channel. Requires membership in the channel.
+channel_uid (string) - Channel UID
+username (string) - Username to invite
+Returns: {success: true, username: "..."}
Leave a channel. Cannot leave the public channel.
+channel_uid (string) - Channel UID
+Returns: {success: true}
Clear all messages from a channel. Admin only.
+channel_uid (string) - Channel UID
+Returns: Boolean success
+Send a message to a channel. If is_final is false, the message can be updated (for typing indicators).
+channel_uid (string) - Target channel UID
+message (string) - Message content
+is_final (boolean, optional) - Whether the message is final (default true)
+Returns: Message UID on success
+Get messages from a channel with pagination.
+channel_uid (string) - Channel UID
+offset (integer, optional) - Pagination offset
+timestamp (string, optional) - Filter by timestamp
+Returns: Array of message objects
+Update a non-final message's text. Only the message author can update. Message must be less than 8 seconds old.
+message_uid (string) - Message UID
+text (string) - New text content
+Returns: {success: true}
Mark a message as final, preventing further updates.
+message_uid (string) - Message UID
+Returns: Boolean success
+Mark all messages in a channel as read for the current user.
+channel_uid (string) - Channel UID
+Returns: Boolean success
+Get the UID of the first unread message in a channel.
+channel_uid (string) - Channel UID
+Returns: Message UID or null
+Broadcast a typing indicator to channel members.
+channel_uid (string) - Channel UID
+color (string, optional) - Indicator color (defaults to user's color)
+Returns: Boolean success
+Get users currently online in a channel.
+channel_uid (string) - Channel UID
+Returns: Array of user objects sorted by nick
+Get users with recent activity in a channel.
+channel_uid (string) - Channel UID
+Returns: Array with uid, username, nick, last_ping
+Get all members of a channel.
+channel_uid (string) - Channel UID
+Returns: Array with uid, username, nick, last_ping
+Container methods provide access to channel-specific Docker containers (Ubuntu terminals).
+ +Get container information for a channel.
+channel_uid (string) - Channel UID
+Returns: Container object with name, cpus, memory, image, status
+Start the container for a channel.
+channel_uid (string) - Channel UID
+Returns: Boolean success
+Stop the container for a channel.
+channel_uid (string) - Channel UID
+Returns: Boolean success
+Get the current status of a channel's container.
+channel_uid (string) - Channel UID
+Returns: Status string
+Write to container stdin and capture output.
+channel_uid (string) - Channel UID
+content (string) - Content to write to stdin
+timeout (integer, optional) - Wait timeout in seconds (max 30)
+Returns: Output string
+Database methods provide direct access to user-scoped database operations.
+ +Insert a record into a user-scoped table.
+table_name (string) - Table name
+record (object) - Record to insert
+Returns: Inserted record
+Update a record in a user-scoped table.
+table_name (string) - Table name
+record (object) - Record with updates
+Returns: Updated record
+Delete a record from a user-scoped table.
+table_name (string) - Table name
+record (object) - Record identifier
+Returns: Boolean success
+Get a single record from a user-scoped table.
+table_name (string) - Table name
+record (object) - Query criteria
+Returns: Record or null
+Find records in a user-scoped table.
+table_name (string) - Table name
+record (object) - Query criteria
+Returns: Array of records
+Insert or update a record based on key fields.
+table_name (string) - Table name
+record (object) - Record data
+keys (array) - Key fields for matching
+Returns: Upserted record
+Execute a raw SQL query on user-scoped data.
+sql (string) - SQL query
+args (array) - Query parameters
+Returns: Query results
+Execute a read-only SQL query. Forbidden keywords (DROP, ALTER, etc.) are blocked.
+sql (string) - SQL SELECT query
+Returns: Array of records (sensitive fields filtered)
+Echo back the provided arguments. Useful for testing.
+args - Any arguments
+Returns: The same arguments
+Send a raw JSON object through the WebSocket. No response is sent back.
+obj (object) - Object to send
+Returns: No response
+Broadcast a stars render event to all online users in a channel.
+channel_uid (string) - Channel UID
+message (string) - Message content
+Returns: Boolean success
+Technical architecture and system design of the Snek platform
+Snek follows a layered architecture pattern with clear separation of concerns. The system is built on Python's aiohttp framework for asynchronous HTTP and WebSocket handling.
+The backend is built on aiohttp, a high-performance asynchronous HTTP client/server framework. Key features:
+Services encapsulate business logic and are accessed via a singleton registry:
+user = await app.services.user.get(uid=user_uid) +channel = await app.services.channel.create(label="#general")+
Available services include: user, channel, channel_member, channel_message, chat, socket, container, db, and more.
+ +Mappers provide data access abstraction with consistent CRUD operations:
+await mapper.get(uid=uid) +await mapper.find(field=value) +await mapper.save(model) +async for record in mapper.query(sql, params): ...+
All mappers support soft delete via the deleted_at field.
The service layer implements UID-based caching for frequently accessed data. Cache invalidation occurs on model updates.
+The frontend uses pure ES6 JavaScript modules without any frameworks. This ensures:
+src/snek/static/ + app.js # Main Application class (singleton) + socket.js # WebSocket RPC client + chat-window.js # Chat interface component + message-list.js # Message rendering + chat-input.js # Input handling + ...+ +
UI components are implemented as Custom Elements:
+<chat-window></chat-window> +<message-list></message-list> +<channel-menu></channel-menu> +<nav-menu></nav-menu>+
Each component has its own CSS file (no Shadow DOM per project guidelines).
+Snek uses Dataset, a simple database abstraction layer built on SQLAlchemy. It provides:
+user # User accounts +channel # Chat channels +channel_member # Channel membership +channel_message# Chat messages +session # User sessions +...+ +
Records are never physically deleted. Instead, the deleted_at timestamp is set. All queries filter by deleted_at IS NULL by default.
Real-time features use a JSON-RPC style protocol over WebSocket:
+{
+ "callId": "1",
+ "method": "send_message",
+ "args": ["channel_uid", "Hello world", true]
+}
+ The server responds with:
+{
+ "callId": "1",
+ "success": true,
+ "data": "message_uid"
+}
+
+ Server-initiated events are broadcast to channel subscribers:
+{
+ "channel_uid": "...",
+ "event": "new_message",
+ "data": { ... }
+}
+
+ Users automatically subscribe to all their channels on WebSocket connection. The socket service manages subscriptions and delivers messages to the appropriate clients.
+src/snek/ + __main__.py # CLI entry point + app.py # Application setup and routing + view/ # HTTP/WebSocket handlers + index.py # Landing page + rpc.py # WebSocket RPC + settings/ # Settings views + service/ # Business logic layer + mapper/ # Data persistence + model/ # Data models + form/ # Form definitions + system/ # Core infrastructure + view.py # BaseView class + service.py # BaseService class + mapper.py # BaseMapper class + model.py # BaseModel class + static/ # Frontend JavaScript and CSS + templates/ # Jinja2 templates+
Build automated bots for the Snek platform
+Snek bots are automated clients that connect via WebSocket and interact with the platform using the RPC API. Bots can:
+Bots authenticate like regular users. Create a dedicated user account for your bot.
+Bots communicate over WebSocket using a JSON-RPC style protocol:
+ +Connect to wss://your-server/rpc.ws (or ws:// for local development).
{
+ "callId": "1",
+ "method": "login",
+ "args": ["bot_username", "bot_password"]
+}
+
+ {
+ "callId": "1",
+ "success": true,
+ "data": { "uid": "...", "username": "bot_username", ... }
+}
+
+ After login, the server pushes events for new messages:
+{
+ "channel_uid": "abc123",
+ "event": "new_message",
+ "data": {
+ "uid": "msg_uid",
+ "message": "Hello bot!",
+ "user_uid": "sender_uid",
+ "username": "sender_name"
+ }
+}
+ A complete, working Python bot using aiohttp:
+ + Python +# retoor <retoor@molodetz.nl>
+
+import asyncio
+import json
+import aiohttp
+
+
+class SnekBot:
+ def __init__(self, base_url, username, password):
+ self.base_url = base_url
+ self.username = username
+ self.password = password
+ self.ws = None
+ self.session = None
+ self.call_id = 0
+ self.pending = {}
+ self.running = True
+
+ async def connect(self):
+ self.session = aiohttp.ClientSession()
+ ws_url = self.base_url.replace("https", "wss").replace("http", "ws")
+ ws_url = f"{ws_url}/rpc.ws"
+ self.ws = await self.session.ws_connect(ws_url, heartbeat=30)
+ asyncio.create_task(self._message_loop())
+ user = await self.login()
+ print(f"Logged in as {user.get('username')}")
+ return user
+
+ async def _message_loop(self):
+ try:
+ async for msg in self.ws:
+ if msg.type == aiohttp.WSMsgType.TEXT:
+ data = json.loads(msg.data)
+ call_id = data.get("callId")
+ if call_id and call_id in self.pending:
+ self.pending[call_id].set_result(data)
+ elif data.get("event") == "new_message":
+ await self.on_message(data.get("data", {}))
+ elif data.get("event"):
+ await self.on_event(data)
+ elif msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR):
+ break
+ except Exception as e:
+ print(f"Message loop error: {e}")
+ finally:
+ self.running = False
+
+ async def call(self, method, *args):
+ self.call_id += 1
+ call_id = str(self.call_id)
+ future = asyncio.get_event_loop().create_future()
+ self.pending[call_id] = future
+ await self.ws.send_json({
+ "callId": call_id,
+ "method": method,
+ "args": list(args)
+ })
+ try:
+ result = await asyncio.wait_for(future, timeout=30)
+ return result.get("data")
+ finally:
+ self.pending.pop(call_id, None)
+
+ async def login(self):
+ return await self.call("login", self.username, self.password)
+
+ async def send_message(self, channel_uid, message):
+ return await self.call("send_message", channel_uid, message, True)
+
+ async def get_channels(self):
+ return await self.call("get_channels")
+
+ async def on_message(self, data):
+ message = data.get("message", "")
+ channel_uid = data.get("channel_uid")
+ username = data.get("username", "")
+
+ if username == self.username:
+ return
+
+ if message.lower().startswith("!ping"):
+ await self.send_message(channel_uid, "Pong!")
+
+ elif message.lower().startswith("!help"):
+ help_text = "Available commands: !ping, !help, !time"
+ await self.send_message(channel_uid, help_text)
+
+ elif message.lower().startswith("!time"):
+ import datetime
+ now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+ await self.send_message(channel_uid, f"Current time: {now}")
+
+ async def on_event(self, data):
+ event = data.get("event")
+ print(f"Event: {event}")
+
+ async def close(self):
+ self.running = False
+ if self.ws:
+ await self.ws.close()
+ if self.session:
+ await self.session.close()
+
+
+async def main():
+ bot = SnekBot(
+ base_url="https://snek.community",
+ username="my_bot",
+ password="my_bot_password"
+ )
+
+ try:
+ await bot.connect()
+ channels = await bot.get_channels()
+ print(f"Member of {len(channels)} channels")
+
+ while bot.running:
+ await asyncio.sleep(1)
+ except KeyboardInterrupt:
+ print("Shutting down...")
+ finally:
+ await bot.close()
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
+ A complete Node.js bot using the ws package:
+ + JavaScript (Node.js) +// retoor <retoor@molodetz.nl>
+
+const WebSocket = require("ws")
+
+class SnekBot {
+ constructor(baseUrl, username, password) {
+ this.baseUrl = baseUrl
+ this.username = username
+ this.password = password
+ this.ws = null
+ this.callId = 0
+ this.pending = new Map()
+ this.running = true
+ }
+
+ connect() {
+ return new Promise((resolve, reject) => {
+ const wsUrl = this.baseUrl
+ .replace("https", "wss")
+ .replace("http", "ws") + "/rpc.ws"
+
+ this.ws = new WebSocket(wsUrl)
+
+ this.ws.on("open", async () => {
+ try {
+ const user = await this.login()
+ console.log(`Logged in as ${user.username}`)
+ resolve(user)
+ } catch (e) {
+ reject(e)
+ }
+ })
+
+ this.ws.on("message", (data) => {
+ const parsed = JSON.parse(data.toString())
+ const callId = parsed.callId
+
+ if (callId && this.pending.has(callId)) {
+ this.pending.get(callId).resolve(parsed)
+ this.pending.delete(callId)
+ } else if (parsed.event === "new_message") {
+ this.onMessage(parsed.data || {})
+ } else if (parsed.event) {
+ this.onEvent(parsed)
+ }
+ })
+
+ this.ws.on("close", () => {
+ this.running = false
+ console.log("Connection closed")
+ })
+
+ this.ws.on("error", (err) => {
+ console.error("WebSocket error:", err.message)
+ reject(err)
+ })
+ })
+ }
+
+ call(method, ...args) {
+ return new Promise((resolve, reject) => {
+ this.callId++
+ const callId = String(this.callId)
+
+ const timeout = setTimeout(() => {
+ this.pending.delete(callId)
+ reject(new Error("Request timeout"))
+ }, 30000)
+
+ this.pending.set(callId, {
+ resolve: (data) => {
+ clearTimeout(timeout)
+ resolve(data.data)
+ },
+ reject
+ })
+
+ this.ws.send(JSON.stringify({
+ callId,
+ method,
+ args
+ }))
+ })
+ }
+
+ login() {
+ return this.call("login", this.username, this.password)
+ }
+
+ sendMessage(channelUid, message) {
+ return this.call("send_message", channelUid, message, true)
+ }
+
+ getChannels() {
+ return this.call("get_channels")
+ }
+
+ async onMessage(data) {
+ const message = data.message || ""
+ const channelUid = data.channel_uid
+ const username = data.username || ""
+
+ if (username === this.username) {
+ return
+ }
+
+ if (message.toLowerCase().startsWith("!ping")) {
+ await this.sendMessage(channelUid, "Pong!")
+ } else if (message.toLowerCase().startsWith("!help")) {
+ await this.sendMessage(channelUid, "Commands: !ping, !help, !time")
+ } else if (message.toLowerCase().startsWith("!time")) {
+ const now = new Date().toISOString()
+ await this.sendMessage(channelUid, `Current time: ${now}`)
+ }
+ }
+
+ onEvent(data) {
+ console.log(`Event: ${data.event}`)
+ }
+
+ close() {
+ this.running = false
+ if (this.ws) {
+ this.ws.close()
+ }
+ }
+}
+
+async function main() {
+ const bot = new SnekBot(
+ "https://snek.community",
+ "my_bot",
+ "my_bot_password"
+ )
+
+ try {
+ await bot.connect()
+ const channels = await bot.getChannels()
+ console.log(`Member of ${channels.length} channels`)
+
+ process.on("SIGINT", () => {
+ console.log("Shutting down...")
+ bot.close()
+ process.exit(0)
+ })
+ } catch (e) {
+ console.error("Failed to connect:", e.message)
+ process.exit(1)
+ }
+}
+
+main()
+ Bots receive various events after authentication:
+ +Received when a message is posted to a channel the bot is a member of.
+{
+ "channel_uid": "...",
+ "event": "new_message",
+ "data": {
+ "uid": "message_uid",
+ "message": "Message content",
+ "user_uid": "sender_uid",
+ "username": "sender_username",
+ "nick": "Sender Nick",
+ "created_at": "2025-01-01T12:00:00"
+ }
+}
+
+ Received when a user starts typing.
+{
+ "channel_uid": "...",
+ "event": "set_typing",
+ "data": {
+ "user_uid": "...",
+ "username": "...",
+ "nick": "...",
+ "color": "#7ef"
+ }
+}
+
+ Received when a message is edited (non-final messages only).
+{
+ "channel_uid": "...",
+ "event": "update_message_text",
+ "data": {
+ "message_uid": "...",
+ "text": "Updated message content"
+ }
+}
+
+ Received when a user's online status changes.
+Implement rate limiting to avoid overwhelming the server:
+Handle errors gracefully:
+Implement automatic reconnection:
+Clean up resources on shutdown:
+Integrate with AI services by calling external APIs when messages match certain patterns:
+async def on_message(self, data):
+ message = data.get("message", "")
+ if message.startswith("!ask "):
+ query = message[5:]
+ response = await self.call_ai_api(query)
+ await self.send_message(data["channel_uid"], response)
+
+ Use asyncio tasks for periodic operations:
+async def daily_report(self): + while self.running: + await asyncio.sleep(86400) # 24 hours + for channel in await self.get_channels(): + await self.send_message(channel["uid"], "Daily report...")+ +
Run multiple bot instances for different purposes. Each bot needs its own user account and WebSocket connection.
+How to contribute to the Snek platform
+git clone https://retoor.molodetz.nl/retoor/snek.git +cd snek+ +
make install+
This creates a virtual environment and installs all dependencies.
+ +make run+
The server starts at http://localhost:8080 by default.
pytest tests/+
Every file must include the author tag at the top:
+# retoor <retoor@molodetz.nl>+
For JavaScript:
+// retoor <retoor@molodetz.nl>+ +
Code should be self-documenting. If you need to explain something, refactor the code to be clearer. Comments indicate code that needs improvement.
+ +The frontend uses vanilla ES6 JavaScript only. No React, Vue, Angular, or similar frameworks. Use Custom Elements for component encapsulation.
+ +Prefer explicit code paths. No magic imports, no hidden side effects, no implicit type conversions.
+ +Use clear, descriptive names. Avoid abbreviations except for common conventions (uid, url, etc.). No suffixes like _super or _refactored.
JavaScript: one class per file. Python: related classes can share a file, but keep files focused. Use directories to group related modules.
+Views handle HTTP requests and render responses.
+ +# retoor <retoor@molodetz.nl>
+
+from snek.system.view import BaseView
+
+
+class MyFeatureView(BaseView):
+ login_required = True
+
+ async def get(self):
+ data = await self.services.my_service.get_data()
+ return await self.render_template("my_feature.html", {"data": data})
+
+ Add a template at templates/my_feature.html:
{% raw %}{% extends "base.html" %}
+
+{% block title %}My Feature{% endblock %}
+
+{% block main %}
+<div class="container">
+ <h1>My Feature</h1>
+ <p>{{ data }}</p>
+</div>
+{% endblock %}{% endraw %}
+
+ In app.py setup_router method:
from snek.view.my_feature import MyFeatureView
+
+self.router.add_view("/my-feature.html", MyFeatureView)
+ Services contain business logic and are accessed via app.services.name.
# retoor <retoor@molodetz.nl>
+
+from snek.system.service import BaseService
+
+
+class MyService(BaseService):
+ mapper_class = None
+
+ async def process_data(self, data):
+ return {"processed": data}
+
+ Services are auto-discovered from the service/ directory. The service name is derived from the filename (e.g., my_service.py becomes app.services.my_service).
RPC methods are defined in the RPCApi class in view/rpc.py.
async def my_method(self, param1, param2=None):
+ self._require_login()
+ self._require_services()
+
+ if not param1 or not isinstance(param1, str):
+ raise ValueError("Invalid param1")
+
+ result = await self.services.my_service.process(param1, param2)
+ return result
+
+ self._require_login() for authenticated methodsself._require_services() to ensure services are availableValueError for invalid inputPermissionError for authorization failuresFrontend components use Custom Elements.
+ +// retoor <retoor@molodetz.nl>
+
+class MyComponent extends HTMLElement {
+ constructor() {
+ super()
+ this.data = null
+ }
+
+ connectedCallback() {
+ this.render()
+ }
+
+ render() {
+ this.innerHTML = `
+ <div class="my-component">
+ <h2>My Component</h2>
+ </div>
+ `
+ }
+}
+
+customElements.define("my-component", MyComponent)
+
+export { MyComponent }
+
+ Add styles in a separate CSS file (no Shadow DOM):
+.my-component {
+ background: #1a1a1a;
+ padding: 1rem;
+ border-radius: 6px;
+}
+
+.my-component h2 {
+ color: #7ef;
+}
+
+ <script type="module" src="/my-component.js"></script> +<link rel="stylesheet" href="/my-component.css">+
pytest tests/ # All tests +pytest tests/test_user.py # Single file +pytest tests/test_user.py::test_login # Single test +pytest -m unit # Unit tests only +pytest -m integration # Integration tests+ +
# retoor <retoor@molodetz.nl> + +import pytest + + +@pytest.mark.unit +async def test_my_feature(): + result = await my_function() + assert result is not None+
mainfeat: add user profile editing +fix: resolve channel subscription issue +docs: update API documentation +refactor: simplify message handling+
Philosophy and technical choices that shape the Snek platform
+Snek is built on a set of guiding principles that inform every technical and design decision:
+Rationale:
+Rationale:
+Rationale:
+Rationale:
+Snek does not record IP addresses, request logs, or user behavior patterns. This is a fundamental privacy commitment, not a configuration option.
+ +Email is optional. Users can register and participate without providing any personally identifiable information. Password recovery relies on administrator assistance.
+ +Sessions use secure, HTTP-only cookies with server-side storage. Session data is encrypted and expires after configurable inactivity periods.
+ +Passwords are hashed using SHA-256 with a configurable salt. The plain password is never stored or logged.
+The dark theme reduces eye strain for extended use, is preferred by developers, and provides better contrast for code display.
+ +Content is organized in cards with consistent padding, border-radius, and shadow. Cards provide visual grouping and work well across screen sizes.
+ +Animations are limited to subtle transitions (hover states, menu opens). No gratuitous motion that could distract or slow down interaction.
+ +Each component has its own CSS file that integrates with the global stylesheet. This allows consistent theming and easier debugging compared to Shadow DOM isolation.
+Code should be self-documenting through clear naming and structure. Comments indicate code that needs refactoring.
+ +Every file includes the author tag at the top:
+# retoor <retoor@molodetz.nl>
+
+ Code is organized into small, focused modules. One class per file for JavaScript. Related functionality grouped in directories.
+ +Prefer explicit code paths over magic behavior. No auto-imports, no implicit conversions, no hidden side effects.
+