python-nostr

NIP-26 allows an 'identity' key (kept offline) to authorize a 'delegatee' key (used in hot clients) to sign specific event kinds for a set duration.

1. Create a Delegation object specifying the delegator_pubkey, delegatee_pubkey, event_kind, and duration_secs.
2. Sign the delegation using the identity key: identity_pk.sign_delegation(delegation).
3. When the delegatee creates an event, include the delegation tag: tags=[delegation.get_tag()].
4. The delegatee then signs the event normally.

Note: Delegations cannot be revoked. It is best practice to use a relatively short duration_secs.

from nostr.delegation import Delegation
from nostr.event import EventKind, Event
from nostr.key import PrivateKey

# Load your "identity" PK that you'd like to keep safely offline
identity_pk = PrivateKey.from_nsec("nsec1...")

# Create a new, disposable PK as the "delegatee" that can be "hot" in a Nostr client
delegatee_pk = PrivateKey()

# the "identity" PK will authorize "delegatee" to sign TEXT_NOTEs on its behalf for the next month
delegation = Delegation(
    delegator_pubkey=identity_pk.public_key.hex(),
    delegatee_pubkey=delegatee_pk.public_key.hex(),
    event_kind=EventKind.TEXT_NOTE,
    duration_secs=30*24*60*60
)

identity_pk.sign_delegation(delegation)

event = Event(
    "Hello, NIP-26!",
    tags=[delegation.get_tag()],
)
delegatee_pk.sign_event(event)

# ...normal broadcast steps...

To listen for specific events (like TEXT_NOTE from specific authors), use Filters and Filter objects.

1. Define a Filter with criteria (e.g., authors, kinds).
2. Wrap filters in a Filters object.
3. Use relay_manager.add_subscription(subscription_id, filters) to register the interest.
4. Use relay_manager.publish_message(json_request) to send the subscription request to the relays.
5. Iterate through relay_manager.message_pool.has_events() and retrieve events using get_event().

import json
import ssl
import time
from nostr.filter import Filter, Filters
from nostr.event import Event, EventKind
from nostr.relay_manager import RelayManager
from nostr.message_type import ClientMessageType

filters = Filters([Filter(authors=[<a nostr pubkey in hex>], kinds=[EventKind.TEXT_NOTE])])
subscription_id = <a string to identify a subscription>
request = [ClientMessageType.REQUEST, subscription_id]
request.extend(filters.to_json_array())

relay_manager = RelayManager()
relay_manager.add_subscription(subscription_id, filters)
relay_manager.open_connections({"cert_reqs": ssl.CERT_NONE})
time.sleep(1.25)

message = json.dumps(request)
relay_manager.publish_message(message)
time.sleep(1)

while relay_manager.message_pool.has_events():
  event_msg = relay_manager.message_pool.get_event()
  print(event_msg.event.content)
  
relay_manager.close_connections()

Use RelayManager to manage connections to multiple WebSocket relays.

1. Instantiate RelayManager.
2. Add relays using .add_relay(url).
3. Open connections using .open_connections(options). Note that passing {"cert_reqs": ssl.CERT_NONE} disables SSL certificate verification.
4. Use time.sleep() to allow connections to stabilize before interacting with the message_pool.
5. Access incoming notices via relay_manager.message_pool.get_notice().

import ssl
import time
from nostr.relay_manager import RelayManager

relay_manager = RelayManager()
relay_manager.add_relay("wss://nostr-pub.wellorder.net")
relay_manager.add_relay("wss://relay.damus.io")
relay_manager.open_connections({"cert_reqs": ssl.CERT_NONE}) # NOTE: This disables ssl certificate verification
time.sleep(1.25) # allow the connections to open

while relay_manager.message_pool.has_notices():
  notice_msg = relay_manager.message_pool.get_notice()
  print(notice_msg.content)
  
relay_manager.close_connections()

You can subscribe to specific event filters on a single relay or across all managed relays.

Subscribe on a specific relay

Use add_subscription_on_relay to target one URL. This will raise a RelayException if the relay is not configured to allow reading (relay.policy.should_read) or if the URL is unknown.

Subscribe on all relays

Use add_subscription_on_all_relays to broadcast a subscription to every relay in the manager that is configured for reading.

Parameters:

• id (str): A unique identifier for the subscription.
• filters (Filters): The Nostr filter object defining what events to listen for.
• url (str): (For single relay) The target relay URL.

from nostr.filter import Filters

filters = Filters(kinds=[1])
# Subscribe on all relays
manager.add_subscription_on_all_relays(id="sub_1", filters=filters)

# Subscribe on a specific relay
manager.add_subscription_on_relay(url="wss://relay.example.com", id="sub_2", filters=filters)

The Relay class is the primary interface for connecting to a Nostr relay, managing subscriptions, and publishing messages. It handles the underlying WebSocket connection, automatic reconnection logic, and message routing to a MessagePool.

Key Methods

• connect(): Establishes the WebSocket connection to the relay URL. It supports SSL options and proxy configurations.
• close(): Gracefully closes the connection.
• publish(message: str): Adds a message to an internal queue to be sent to the relay via a background worker.
• add_subscription(id, filters: Filters): Registers a new subscription with a unique ID and specific event filters.
• close_subscription(id: str): Removes an existing subscription by its ID.
• update_subscription(id: str, filters: Filters): Updates the filters for an existing subscription ID.

Configuration

When initializing a Relay, you can provide:

• url: The WebSocket URL of the relay.
• message_pool: An instance of MessagePool where incoming messages are stored.
• policy: A RelayPolicy object to control reading/writing permissions.
• ssl_options: A dictionary for SSL configuration.
• proxy_config: A RelayProxyConnectionConfig object for proxying connections.

from nostr.relay import Relay
from nostr.message_pool import MessagePool
from nostr.filter import Filters

# Initialize components
pool = MessagePool()
filters = Filters(...) # Define your filters

# Create and connect the relay
relay = Relay(url="wss://relay.example.com", message_pool=pool)
relay.connect()

# Manage subscriptions
relay.add_subscription("sub_1", filters)

# Publish a message
relay.publish('["EVENT", "sub_1", {...}]')