pulse-voice 1.0.1

Official Python library for the Pulse API - Build bots and integrations

Pulse.py

Official Python library for the Pulse API. Build bots and integrations for the Pulse platform.

Installation

pip install pulse-voice

Quick Start

import asyncio
from pulse import Client

client = Client(debug=True)

@client.event
async def on_ready():
    print(f"Logged in as {client.user.username}")

@client.event
async def on_message(data):
    message = data["message"]
    author = data["author"]

    if message["content"] == "!ping":
        await client.rest.send_message(message["roomId"], "Pong!")

    if message["content"] == "!hello":
        await client.rest.send_message(
            message["roomId"],
            f"Hello {author['displayName']}!"
        )

asyncio.run(client.run("YOUR_BOT_TOKEN"))

Features

• Async/Await - Fully asynchronous with asyncio
• REST API Client - Full access to the Pulse API
• WebSocket Events - Real-time message and presence updates
• Type Hints - Full type annotations included
• Easy to Use - Simple and Pythonic API

Documentation

Client

The main client for interacting with Pulse.

from pulse import Client

client = Client(
    api_url="https://pulse.shadowforge.fr/api",  # Optional
    ws_url="wss://pulse.shadowforge.fr",         # Optional
    debug=False,                                  # Optional
)

Events

Use the @client.event decorator or @client.on("event_name"):

@client.event
async def on_ready():
    print("Bot is ready!")

@client.event
async def on_message(data):
    print(f"Message: {data['message']['content']}")

@client.on("member_join")
async def handle_join(data):
    print(f"New member joined!")

Event	Description
ready	Fired when the client is connected and ready
message	Fired when a message is received
message_update	Fired when a message is edited
message_delete	Fired when a message is deleted
typing_start	Fired when someone starts typing
typing_stop	Fired when someone stops typing
presence_update	Fired when a user's presence changes
member_join	Fired when a member joins a hub
member_leave	Fired when a member leaves a hub
error	Fired when an error occurs
disconnect	Fired when disconnected from WebSocket

REST API

Access the REST API through client.rest:

# Get current user
user = await client.rest.get_current_user()

# Send a message
await client.rest.send_message(room_id, "Hello!")

# Get hub members
members = await client.rest.get_hub_members(hub_id)

# Create a room
room = await client.rest.create_room(
    hub_id,
    name="general",
    type="text"
)

WebSocket

Direct WebSocket access through client.ws:

# Join a room for updates
await client.ws.join_room(room_id)

# Leave a room
await client.ws.leave_room(room_id)

# Send typing indicator
await client.ws.start_typing(room_id)
await client.ws.stop_typing(room_id)

Example Bot

import asyncio
import os
from pulse import Client

client = Client(debug=True)
PREFIX = "!"

@client.event
async def on_ready():
    print(f"Bot is ready! Logged in as {client.user.username}")
    print(f"Serving {len(client.hubs)} hubs")

@client.event
async def on_message(data):
    message = data["message"]
    author = data["author"]

    # Ignore self
    if author["id"] == client.user.id:
        return

    content = message["content"]
    room_id = message["roomId"]

    if not content.startswith(PREFIX):
        return

    args = content[len(PREFIX):].strip().split()
    command = args.pop(0).lower() if args else ""

    if command == "ping":
        await client.rest.send_message(room_id, "Pong!")

    elif command == "userinfo":
        info = f"**User Info**\n"
        info += f"Username: {author['username']}\n"
        info += f"ID: {author['id']}\n"
        info += f"Created: {author['createdAt']}"
        await client.rest.send_message(room_id, info)

    elif command == "servercount":
        await client.rest.send_message(
            room_id,
            f"I'm in {len(client.hubs)} hubs!"
        )

    elif command == "echo":
        text = " ".join(args) if args else "Nothing to echo!"
        await client.rest.send_message(room_id, text)

@client.event
async def on_member_join(data):
    print(f"New member joined: {data}")

@client.event
async def on_error(error):
    print(f"Error: {error}")

if __name__ == "__main__":
    token = os.getenv("BOT_TOKEN")
    if not token:
        print("Please set BOT_TOKEN environment variable")
        exit(1)

    asyncio.run(client.run(token))

Links

• Documentation
• Developer Portal
• GitHub
• Discord

License

MIT License - Shadow Forge