diff --git a/scripts/publish_provider.py b/scripts/publish_provider.py new file mode 100644 index 00000000..a51e80e4 --- /dev/null +++ b/scripts/publish_provider.py @@ -0,0 +1,268 @@ +#!/usr/bin/env python3 +""" +Simple Python function to publish one provider listing to a nostr relay +according to the RIP-02 specification. + +Based on: https://github.com/Routstr/protocol/blob/main/RIP-02.md +Event Kind: 31338 (Routstr Provider Announcements) +""" + +import hashlib +import json +import time +import asyncio +from typing import Any + +import secp256k1 +import websockets + + +def create_provider_announcement_event( + private_key_hex: str, + provider_name: str, + endpoint_url: str, + d_tag: str, + description: str | None = None, + contact: str | None = None, + pricing_url: str | None = None, + supported_models: list[str] | None = None, +) -> dict[str, Any]: + """ + Create a RIP-02 compliant provider announcement event. + + Args: + private_key_hex: 32-byte hex private key for signing + provider_name: Human readable name for the provider + endpoint_url: Base URL for the provider's API endpoint + d_tag: Unique identifier for this provider (required for addressable events) + description: Optional description of the provider + contact: Optional contact information + pricing_url: Optional URL to pricing information + supported_models: Optional list of supported model names + + Returns: + Complete signed nostr event ready for publishing + """ + # Convert hex private key to secp256k1 PrivateKey object + private_key = secp256k1.PrivateKey(bytes.fromhex(private_key_hex)) + public_key = private_key.pubkey.serialize(compressed=True)[ + 1: + ] # Remove 0x02/0x03 prefix + + # Build required tags according to RIP-02 + tags = [ + ["d", d_tag], # Required for addressable events (kind 30000-39999) + ["endpoint", endpoint_url], + ["name", provider_name], + ] + + # Add optional tags if provided + if description: + tags.append(["description", description]) + if contact: + tags.append(["contact", contact]) + if pricing_url: + tags.append(["pricing", pricing_url]) + if supported_models: + for model in supported_models: + tags.append(["model", model]) + + # Create the event structure + created_at = int(time.time()) + event_data = [ + 0, # Reserved field + public_key.hex(), # Public key as hex + created_at, # Unix timestamp + 31338, # Kind for RIP-02 Provider Announcements + tags, # Tags array + "", # Content (empty for provider announcements) + ] + + # Serialize event data for hashing + event_json = json.dumps(event_data, separators=(",", ":"), ensure_ascii=False) + + # Calculate event ID (SHA256 hash) + event_id = hashlib.sha256(event_json.encode("utf-8")).hexdigest() + + # Sign the event ID + signature = private_key.ecdsa_sign(bytes.fromhex(event_id), raw=True) + signature_der = private_key.ecdsa_serialize(signature) + + # Create the final event + event = { + "id": event_id, + "pubkey": public_key.hex(), + "created_at": created_at, + "kind": 31338, + "tags": tags, + "content": "", + "sig": signature_der.hex(), + } + + return event + + +async def publish_provider_to_relay( + relay_url: str, event: dict[str, Any], timeout: int = 30 +) -> bool: + """ + Publish a provider announcement event to a nostr relay. + + Args: + relay_url: WebSocket URL of the nostr relay (e.g., "wss://relay.damus.io") + event: Complete signed nostr event to publish + timeout: Connection timeout in seconds + + Returns: + True if successfully published, False otherwise + """ + try: + async with websockets.connect(relay_url, timeout=timeout) as websocket: + # Send EVENT message + event_message = json.dumps(["EVENT", event]) + await websocket.send(event_message) + print(f"Published event {event['id']} to {relay_url}") + + # Wait for OK response + try: + response = await asyncio.wait_for(websocket.recv(), timeout=5) + data = json.loads(response) + + if data[0] == "OK" and data[1] == event["id"]: + if data[2]: # True means accepted + print( + f"āœ… Event accepted by relay: {data[3] if len(data) > 3 else ''}" + ) + return True + else: + print( + f"āŒ Event rejected by relay: {data[3] if len(data) > 3 else ''}" + ) + return False + elif data[0] == "NOTICE": + print(f"šŸ“¢ Relay notice: {data[1]}") + return False + else: + print(f"šŸ¤” Unexpected response: {data}") + return False + + except asyncio.TimeoutError: + print("ā° No response from relay within timeout") + return False + + except Exception as e: + print(f"šŸ’„ Failed to publish to {relay_url}: {e}") + return False + + +async def publish_provider_listing( + private_key_hex: str, + provider_name: str, + endpoint_url: str, + d_tag: str, + relay_urls: list[str] | None = None, + description: str | None = None, + contact: str | None = None, + pricing_url: str | None = None, + supported_models: list[str] | None = None, +) -> dict[str, bool]: + """ + Complete function to create and publish a provider listing to nostr relays. + + Args: + private_key_hex: 32-byte hex private key for signing + provider_name: Human readable name for the provider + endpoint_url: Base URL for the provider's API endpoint + d_tag: Unique identifier for this provider + relay_urls: List of relay URLs to publish to (uses defaults if None) + description: Optional description of the provider + contact: Optional contact information + pricing_url: Optional URL to pricing information + supported_models: Optional list of supported model names + + Returns: + Dictionary mapping relay URLs to success status + """ + # Use default relays if none provided + if relay_urls is None: + relay_urls = [ + "wss://relay.nostr.band", + "wss://relay.damus.io", + "wss://relay.routstr.com", + ] + + # Create the provider announcement event + event = create_provider_announcement_event( + private_key_hex=private_key_hex, + provider_name=provider_name, + endpoint_url=endpoint_url, + d_tag=d_tag, + description=description, + contact=contact, + pricing_url=pricing_url, + supported_models=supported_models, + ) + + print(f"šŸ“ Created provider announcement event: {event['id']}") + print(f"šŸ”‘ Public key: {event['pubkey']}") + print(f"šŸ·ļø Provider: {provider_name}") + print(f"🌐 Endpoint: {endpoint_url}") + print() + + # Publish to all specified relays + results = {} + tasks = [] + + for relay_url in relay_urls: + task = publish_provider_to_relay(relay_url, event) + tasks.append((relay_url, task)) + + # Execute all publishing tasks concurrently + for relay_url, task in tasks: + try: + success = await task + results[relay_url] = success + except Exception as e: + print(f"šŸ’„ Failed to publish to {relay_url}: {e}") + results[relay_url] = False + + return results + + +# Example usage +async def main() -> None: + """Example of how to use the provider publishing function.""" + + # Example private key (DO NOT use this in production!) + private_key = "3185a47e3802f956ca207b46c8d6b8b5c5dbad53a5ca29816050e9b66badc33c" + + # Example provider information + provider_name = "My AI Provider" + endpoint_url = "https://api.myaiprovider.com" + d_tag = "my-ai-provider-v1" # Unique identifier + description = "High-quality AI models with competitive pricing" + contact = "admin@myaiprovider.com" + pricing_url = "https://myaiprovider.com/pricing" + supported_models = ["gpt-4o", "claude-3-sonnet", "llama-3.1-70b"] + + # Publish to relays + results = await publish_provider_listing( + private_key_hex=private_key, + provider_name=provider_name, + endpoint_url=endpoint_url, + d_tag=d_tag, + description=description, + contact=contact, + pricing_url=pricing_url, + supported_models=supported_models, + ) + + # Print results + print("\nšŸ“Š Publishing Results:") + for relay_url, success in results.items(): + status = "āœ… Success" if success else "āŒ Failed" + print(f" {relay_url}: {status}") + + +if __name__ == "__main__": + asyncio.run(main())