tools.py

"""Hermes async tools for nostr-profile.

Identity for publish/update is loaded from NOSTRKEY_IDENTITY_FILE +
NOSTRKEY_PASSPHRASE env vars (or NOSTRKEY_NSEC for an unencrypted nsec)
so signing keys never enter the LLM context.
"""

from __future__ import annotations

import os
from typing import Any

from nostrkey import Identity
from nostrkey.keys import npub_to_hex, nsec_to_hex
from nostr_profile import Profile, get_profile, publish_profile, update_profile

from tools.registry import tool_error, tool_result


def _load_identity() -> Identity | None:
    """Load identity from env. Prefer encrypted file; fall back to NOSTRKEY_NSEC."""
    nsec = os.environ.get("NOSTRKEY_NSEC", "").strip()
    if nsec:
        return Identity.from_nsec(nsec)
    file_path = os.environ.get("NOSTRKEY_IDENTITY_FILE", "").strip()
    passphrase = os.environ.get("NOSTRKEY_PASSPHRASE", "")
    if file_path and passphrase:
        return Identity.load(file_path, passphrase)
    return None


def _need_identity() -> str:
    return tool_error(
        "No identity available. Set NOSTRKEY_NSEC, or NOSTRKEY_IDENTITY_FILE + "
        "NOSTRKEY_PASSPHRASE, in the Hermes environment so this tool can sign."
    )


def _resolve_pubkey(npub_or_hex: str) -> str:
    s = (npub_or_hex or "").strip()
    if s.startswith("npub1"):
        return npub_to_hex(s)
    return s


# -----------------------------
# profile_publish
# -----------------------------

PROFILE_PUBLISH_SCHEMA = {
    "type": "function",
    "function": {
        "name": "profile_publish",
        "description": (
            "Publish a complete Nostr profile (kind 0) to a relay. Replaces "
            "any existing profile for this identity. Requires identity env vars."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "relay_url": {"type": "string", "description": "wss:// relay URL."},
                "name":      {"type": "string", "description": "Display name."},
                "about":     {"type": "string", "description": "Bio/description."},
                "picture":   {"type": "string", "description": "URL to avatar image."},
                "banner":    {"type": "string", "description": "URL to banner image."},
                "nip05":     {"type": "string", "description": "NIP-05 verification (e.g. user@domain.com)."},
                "lud16":     {"type": "string", "description": "Lightning address for tips."},
                "website":   {"type": "string", "description": "Personal/agent website URL."},
            },
            "required": ["relay_url"],
        },
    },
}


async def handle_profile_publish(args: dict[str, Any], **kw) -> str:
    identity = _load_identity()
    if identity is None:
        return _need_identity()
    relay_url = args.get("relay_url")
    if not relay_url:
        return tool_error("relay_url is required")
    try:
        profile = Profile(
            name=args.get("name"),
            about=args.get("about"),
            picture=args.get("picture"),
            banner=args.get("banner"),
            nip05=args.get("nip05"),
            lud16=args.get("lud16"),
            website=args.get("website"),
        )
        event_id = await publish_profile(identity, profile, relay_url)
        return tool_result({"event_id": event_id, "npub": identity.npub, "relay_url": relay_url})
    except Exception as e:
        return tool_error(f"profile_publish failed: {type(e).__name__}: {e}")


# -----------------------------
# profile_update
# -----------------------------

PROFILE_UPDATE_SCHEMA = {
    "type": "function",
    "function": {
        "name": "profile_update",
        "description": (
            "Patch fields on an existing Nostr profile. Reads current "
            "profile from the relay, applies non-null fields, and republishes. "
            "Requires identity env vars."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "relay_url": {"type": "string"},
                "name":      {"type": "string"},
                "about":     {"type": "string"},
                "picture":   {"type": "string"},
                "banner":    {"type": "string"},
                "nip05":     {"type": "string"},
                "lud16":     {"type": "string"},
                "website":   {"type": "string"},
            },
            "required": ["relay_url"],
        },
    },
}


async def handle_profile_update(args: dict[str, Any], **kw) -> str:
    identity = _load_identity()
    if identity is None:
        return _need_identity()
    relay_url = args.get("relay_url")
    if not relay_url:
        return tool_error("relay_url is required")
    try:
        kwargs = {k: args[k] for k in ("name", "about", "picture", "banner", "nip05", "lud16", "website") if args.get(k) is not None}
        event_id = await update_profile(identity, relay_url, **kwargs)
        return tool_result({"event_id": event_id, "npub": identity.npub, "updated_fields": list(kwargs.keys())})
    except Exception as e:
        return tool_error(f"profile_update failed: {type(e).__name__}: {e}")


# -----------------------------
# profile_read
# -----------------------------

PROFILE_READ_SCHEMA = {
    "type": "function",
    "function": {
        "name": "profile_read",
        "description": (
            "Fetch a Nostr profile (kind 0 metadata) from a relay by npub or "
            "hex pubkey. No authentication needed."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "pubkey":    {"type": "string", "description": "npub1... or hex public key."},
                "relay_url": {"type": "string", "description": "wss:// relay URL."},
            },
            "required": ["pubkey", "relay_url"],
        },
    },
}


async def handle_profile_read(args: dict[str, Any], **kw) -> str:
    pubkey = args.get("pubkey")
    relay_url = args.get("relay_url")
    if not pubkey or not relay_url:
        return tool_error("pubkey and relay_url are required")
    try:
        pubkey_hex = _resolve_pubkey(pubkey)
        profile = await get_profile(pubkey_hex, relay_url)
        if profile is None:
            return tool_result({"found": False, "pubkey_hex": pubkey_hex})
        return tool_result({"found": True, "pubkey_hex": pubkey_hex, "profile": profile.to_metadata()})
    except Exception as e:
        return tool_error(f"profile_read failed: {type(e).__name__}: {e}")