Post Notes to Nostr Relays

Overview

Publish signed events to Nostr relays. Relays are the message-passing infrastructure of Nostr - you send events to relays, and they distribute them to subscribers.

The Code

"""
Nostr Relay Publisher
Post events to Nostr relays via WebSocket

Requirements: websocket-client (pip install websocket-client)

Note: For production, use nostr-sdk or pynostr for proper event signing.
This example assumes you have a pre-signed event.
"""

import json
import time
import hashlib
import secrets
from typing import List, Optional
import websocket

# Default relays (use your preferred relays)
DEFAULT_RELAYS = [
    "wss://relay.damus.io",
    "wss://nos.lol",
    "wss://relay.nostr.band",
    "wss://nostr.wine"
]


def create_unsigned_event(
    pubkey: str,
    kind: int,
    content: str,
    tags: Optional[List[List[str]]] = None
) -> dict:
    """
    Create an unsigned event structure.
    In production, sign this with your private key.
    """
    if tags is None:
        tags = []

    created_at = int(time.time())

    event = {
        "pubkey": pubkey,
        "created_at": created_at,
        "kind": kind,
        "tags": tags,
        "content": content
    }

    # Calculate event ID
    serialized = json.dumps([
        0, event["pubkey"], event["created_at"],
        event["kind"], event["tags"], event["content"]
    ], separators=(',', ':'), ensure_ascii=False)

    event["id"] = hashlib.sha256(serialized.encode()).hexdigest()

    return event


def publish_to_relay(relay_url: str, event: dict, timeout: int = 10) -> dict:
    """
    Publish an event to a single relay.

    Args:
        relay_url: WebSocket URL of the relay
        event: Signed Nostr event
        timeout: Connection timeout in seconds

    Returns:
        dict with success status and message
    """
    result = {
        "relay": relay_url,
        "success": False,
        "message": ""
    }

    try:
        ws = websocket.create_connection(
            relay_url,
            timeout=timeout
        )

        # Send EVENT message
        message = json.dumps(["EVENT", event])
        ws.send(message)

        # Wait for OK response
        response = ws.recv()
        ws.close()

        data = json.loads(response)

        if data[0] == "OK":
            event_id = data[1]
            accepted = data[2]
            reason = data[3] if len(data) > 3 else ""

            result["success"] = accepted
            result["message"] = reason if reason else "Accepted"
            result["event_id"] = event_id
        else:
            result["message"] = f"Unexpected response: {data[0]}"

    except websocket.WebSocketTimeoutException:
        result["message"] = "Connection timeout"
    except websocket.WebSocketConnectionClosedException:
        result["message"] = "Connection closed"
    except Exception as e:
        result["message"] = str(e)

    return result


def publish_to_relays(
    event: dict,
    relays: Optional[List[str]] = None,
    min_success: int = 1
) -> dict:
    """
    Publish an event to multiple relays.

    Args:
        event: Signed Nostr event
        relays: List of relay URLs (uses defaults if None)
        min_success: Minimum successful publishes required

    Returns:
        dict with overall status and per-relay results
    """
    if relays is None:
        relays = DEFAULT_RELAYS

    results = []
    success_count = 0

    for relay in relays:
        print(f"  Publishing to {relay}...", end=" ", flush=True)
        result = publish_to_relay(relay, event)
        results.append(result)

        if result["success"]:
            success_count += 1
            print(f"OK")
        else:
            print(f"FAILED: {result['message']}")

    return {
        "success": success_count >= min_success,
        "success_count": success_count,
        "total_relays": len(relays),
        "results": results,
        "event_id": event["id"]
    }


def create_text_note(content: str, hashtags: Optional[List[str]] = None) -> dict:
    """
    Create a kind:1 text note event (unsigned).

    Args:
        content: Note text
        hashtags: Optional list of hashtags

    Returns:
        Unsigned event (needs signature)
    """
    tags = []
    if hashtags:
        for tag in hashtags:
            tags.append(["t", tag.lower().strip("#")])

    # Placeholder pubkey - replace with your actual public key
    pubkey = "0" * 64

    return create_unsigned_event(
        pubkey=pubkey,
        kind=1,
        content=content,
        tags=tags
    )


def simulate_signed_event() -> dict:
    """
    Create a simulated signed event for testing.
    In production, use proper key signing.
    """
    event = create_text_note(
        content="Hello from an AI agent! Testing Nostr publishing.",
        hashtags=["nostr", "ai", "agent"]
    )

    # Add a fake signature (won't be accepted by real relays)
    event["sig"] = "0" * 128

    return event


# Example usage
if __name__ == "__main__":
    print("=== Nostr Note Publisher ===\n")

    # Create a test event
    # NOTE: This uses a placeholder signature and won't be accepted
    # In production, sign with your private key
    event = simulate_signed_event()

    print(f"Event ID: {event['id'][:16]}...")
    print(f"Content:  {event['content']}")
    print(f"Tags:     {event['tags']}")

    print(f"\n=== Publishing to Relays ===")

    # Test with a subset of relays
    test_relays = [
        "wss://relay.damus.io",
        "wss://nos.lol"
    ]

    result = publish_to_relays(
        event=event,
        relays=test_relays,
        min_success=1
    )

    print(f"\n=== Results ===")
    print(f"Success: {result['success']}")
    print(f"Accepted by: {result['success_count']}/{result['total_relays']} relays")

    # Note about unsigned events
    print("\n" + "=" * 50)
    print("NOTE: This demo uses unsigned events which relays")
    print("will reject. For real publishing, you need to:")
    print("1. Generate or load your private key")
    print("2. Sign events with BIP-340 Schnorr signatures")
    print("3. Use nostr-sdk or pynostr for production")
    print("=" * 50)

Usage

# Install WebSocket client
pip install websocket-client

# Run the publisher
python post_note.py

Example Output

=== Nostr Note Publisher ===

Event ID: abc123def456...
Content:  Hello from an AI agent! Testing Nostr publishing.
Tags:     [['t', 'nostr'], ['t', 'ai'], ['t', 'agent']]

=== Publishing to Relays ===
  Publishing to wss://relay.damus.io... FAILED: invalid: bad signature
  Publishing to wss://nos.lol... FAILED: invalid: bad signature

=== Results ===
Success: False
Accepted by: 0/2 relays

==================================================
NOTE: This demo uses unsigned events which relays
will reject. For real publishing, you need to:
1. Generate or load your private key
2. Sign events with BIP-340 Schnorr signatures
3. Use nostr-sdk or pynostr for production
==================================================

Agent Notes

Relay selection strategy:

Use multiple relays for redundancy (3-5 recommended)
Include popular relays for reach (damus.io, nos.lol)
Include specialized relays for your use case
Respect relay policies (some require payment or NIP-05)

Publishing best practices:

Publish to multiple relays simultaneously
Consider at least 1-2 successful publishes as “posted”
Retry failed relays with backoff
Store relay results for debugging

Rate limits: Most relays have rate limits. For agents:

Don’t spam (>1 post/minute may trigger limits)
Use reasonable delays between posts
Some relays require proof-of-work (NIP-13)