Relay

This module provides the core relay functionality for the Agentstr SDK, enabling communication between Nostr clients and agents. It handles message routing, event forwarding, and connection management to ensure reliable message delivery across the Nostr network.

pydantic model agentstr.relay.DecryptedMessage

Bases: BaseModel

A decrypted message from a Nostr relay.

Show JSON schema

{
   "title": "DecryptedMessage",
   "description": "A decrypted message from a Nostr relay.",
   "type": "object",
   "properties": {
      "event": {
         "$ref": "#/$defs/Event"
      },
      "message": {
         "title": "Message",
         "type": "string"
      }
   },
   "$defs": {
      "Event": {
         "properties": {
            "content": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Content"
            },
            "pubkey": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Pubkey"
            },
            "created_at": {
               "anyOf": [
                  {
                     "type": "integer"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Created At"
            },
            "kind": {
               "anyOf": [
                  {
                     "type": "integer"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": 1,
               "title": "Kind"
            },
            "tags": {
               "default": null,
               "items": {
                  "items": {
                     "type": "string"
                  },
                  "type": "array"
               },
               "title": "Tags",
               "type": "array"
            },
            "id": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Id"
            },
            "sig": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Sig"
            }
         },
         "title": "Event",
         "type": "object"
      }
   },
   "required": [
      "event",
      "message"
   ]
}

Fields:

• event (pynostr.event.Event)

• message (str)

field event: Event [Required]

The Nostr event containing the message.

field message: str [Required]

The decrypted message content.

agentstr.relay.create_subscription(filters: Filters) → list[str]

Create a subscription for the given filters.

Parameters:

filters – The filters to apply to the subscription.

Returns:

A list containing the subscription request components.

class agentstr.relay.EventRelay(relay: str, private_key: PrivateKey | None = None, public_key: PublicKey | None = None)

Bases: object

Handles communication with a single Nostr relay.

Parameters:

• relay – WebSocket URL of the Nostr relay.

• private_key – Private key for signing events.

• public_key – Optional public key (derived from private_key if not provided).

__init__(relay: str, private_key: PrivateKey | None = None, public_key: PublicKey | None = None)

async get_events(filters: Filters, limit: int = 10, timeout: int = 30, close_on_eose: bool = True) → list[Event]

Fetch events matching the given filters from this relay.

Parameters:

• filters – The filters to apply when fetching events.

• limit – Maximum number of events to return. Defaults to 10.

• timeout – Maximum time to wait for events in seconds. Defaults to 30.

• close_on_eose – Whether to close the subscription after EOSE. Defaults to True.

Returns:

A list of up to limit events that match the filters, or an empty list if none found.

Note

Times out after timeout seconds if no matching events are found.

async get_event(filters: Filters, timeout: int = 30, close_on_eose: bool = True) → Event | None

Get a single event matching the filters or None if not found.

async send_event(event: Event)

Publish an event to this relay.

decrypt_message(event: Event) → DecryptedMessage | None

async send_message(message: str | dict, recipient_pubkey: str, event_ref: str | None = None) → Event

async receive_message(author_pubkey: str, timestamp: int | None = None, timeout: int = 30) → DecryptedMessage | None

Wait for and return the next direct message from the specified author.

async send_receive_message(message: str | dict, recipient_pubkey: str, timeout: int = 3, event_ref: str | None = None) → DecryptedMessage | None

async event_listener(filters: Filters, callback: Callable[[Event], None], event_cache: ExpiringDict, lock: Lock)

Continuously listen for events matching filters and call the callback for each one.

async direct_message_listener(filters: Filters, callback: Callable[[Event, str], None], event_cache: ExpiringDict, lock: Lock)

Listen for direct messages and call the callback with decrypted content.