server.py

"""taxsort-mcp — TaxSort Tollbooth MCP Server.

A monetized MCP server for personal tax transaction storage.
The BE manages raw source data and classifications; AI processing
happens client-side. Standard DPYC tools (check_balance,
purchase_credits, Secure Courier, Oracle, pricing, constraints)
are provided by ``register_standard_tools`` from the tollbooth-dpyc
wheel. Only domain-specific tools are defined here.
"""

from __future__ import annotations

import logging
from typing import Annotated, Any

from fastmcp import FastMCP
from pydantic import Field
from tollbooth.credential_templates import CredentialTemplate, FieldSpec
from tollbooth.credential_validators import validate_btcpay_creds, validate_required
from tollbooth.runtime import OperatorRuntime, register_standard_tools
from tollbooth.tool_identity import STANDARD_IDENTITIES, ToolIdentity, capability_uuid
from tollbooth.version import resolve_service_version


def _validate_taxsort_creds(creds: dict[str, str]) -> list[str]:
    """Validate TaxSort operator credentials."""
    errors = validate_btcpay_creds(creds)
    err = validate_required(creds.get("anthropic_api_key", ""), "anthropic_api_key")
    if err:
        errors.append(err)
    return errors

logger = logging.getLogger(__name__)

__version__ = resolve_service_version("taxsort-mcp", __file__)

# ---------------------------------------------------------------------------
# FastMCP app + slug decorator
# ---------------------------------------------------------------------------

mcp = FastMCP(
    "taxsort-mcp",
    instructions=(
        "TaxSort MCP — Personal tax transaction storage, monetized "
        "via Tollbooth DPYC Bitcoin Lightning micropayments.\n\n"
        "## Onboarding\n"
        "1. Call taxsort_request_npub_proof(patron_npub=...) to start identity verification\n"
        "2. Reply to the Nostr DM with any passphrase to prove npub ownership\n"
        "3. Call taxsort_receive_npub_proof(patron_npub=...) to complete verification\n\n"
        "## Workflow\n"
        "1. taxsort_create_session() → get a session_id\n"
        "2. taxsort_import_csv(session_id, content, filename) → parse and store raw transactions\n"
        "3. taxsort_get_transactions(session_id, unclassified_only=true) → fetch pages for FE classification\n"
        "4. taxsort_save_classifications(session_id, classifications=[...]) → write back AI/manual results\n"
        "5. taxsort_get_transactions(session_id) → review classified results\n"
        "6. taxsort_get_summary(session_id, group_by='taxline') → IRS line totals\n"
        "8. taxsort_create_share_token(session_id) → share with spouse\n\n"
        "## Pricing\n"
        "Tool prices are set dynamically by the operator's pricing model. "
        "Use `taxsort_check_price` to preview costs."
    ),
)

# Shared npub field annotation
NpubField = Annotated[
    str,
    Field(
        description="Required. Your Nostr public key (npub1...) "
        "for credit billing."
    ),
]

# ---------------------------------------------------------------------------
# Tool registry (domain tools only)
# ---------------------------------------------------------------------------

# Frozen UUIDs — declared once at tool birth and never changed.
VERIFY_PASSPHRASE_UUID           = "4e7acfb3-5aee-5046-a882-34730889f0dc"
CREATE_SESSION_UUID              = "9f641066-b0f2-5242-8f96-df7331460bcb"
GET_SESSION_UUID                 = "0b99a03b-62d6-5257-bb85-9e465b70c5e2"
LIST_SESSIONS_UUID               = "f418b6e2-49c3-5297-ab2c-113f8ac0149c"
GET_RULES_UUID                   = "0046328b-0def-585a-8471-736983650450"
COUNT_RULE_MATCHES_UUID          = "474f6a49-5372-507b-a918-f2897ffca8d1"
GET_IMPORT_STATS_UUID            = "a9264a4a-6cd3-5165-b54b-11f9640e3a3b"
LOAD_SHARE_TOKEN_UUID            = "91194142-289e-5f74-ab6d-e03650cc9f9a"
GET_TRANSACTIONS_UUID            = "1404e24a-2a24-504c-93e6-9145ea648b9d"
GET_TRANSACTIONS_PAGED_UUID      = "3389f7e6-09b4-52e0-b7d4-fe5f2041a7af"
GET_SUMMARY_UUID                 = "cd41bd76-d133-5037-a737-0d8889cf3048"
IMPORT_CSV_UUID                  = "8652c842-ae53-5752-8472-8b7f4dc7dd0c"
SAVE_CLASSIFICATIONS_UUID        = "a69c78b5-1d9f-571e-b73b-b008664213f2"
DELETE_CLASSIFICATION_UUID       = "acb46750-74b4-5aca-9306-37e3f9259056"
RESET_CLASSIFICATIONS_UUID       = "a756aaff-1b81-5abd-84f7-5e5b27c91636"
DELETE_ACCOUNT_TRANSACTIONS_UUID = "8390cadc-fc3b-5667-a349-40a658b882c3"
CLEAR_TRANSACTIONS_UUID          = "eafa5cf9-7e90-5905-81d8-c4ff7eaa3763"
GET_AMOUNT_NEIGHBORS_UUID        = "01522e3b-d18e-5a57-b127-36720fe6dfe6"
GET_ACCOUNTS_UUID                = "85599061-3089-5ceb-ba98-fb12022239c6"
SET_ACCOUNT_TYPE_UUID            = "0395b12d-882d-50d3-88ff-e24fe468d804"
SAVE_RULE_UUID                   = "7b554d36-1399-5234-a982-dd6675410bc4"
DELETE_RULE_UUID                 = "04bf01ce-9a00-51b7-848c-daea7b02640f"
APPLY_RULES_UUID                 = "2cb67664-ce3b-58fa-a651-258107d3a62f"
GET_CUSTOM_CATEGORIES_UUID       = "918f91b0-03ef-5ea2-ab67-34467b053057"
SAVE_CUSTOM_CATEGORY_UUID        = "7e3223c9-8f57-5c8d-8bc5-1cb2d32244e3"
DELETE_CUSTOM_CATEGORY_UUID      = "bc859a2c-4fef-54da-b0e1-45183a939caf"
CREATE_SHARE_TOKEN_UUID          = "9c250ace-7ea9-5495-93d4-df9c263746d3"
REQUEST_UNLOCK_UUID              = "9c49b16d-304c-547c-9738-544be684a29d"
CHECK_UNLOCK_UUID                = "0a31531a-0bac-5944-ae0b-686f50d03bb0"
GET_GITHUB_TOKEN_UUID            = "5532f9d6-a465-5f12-a189-896854f1db68"
CREATE_FEEDBACK_ISSUE_UUID       = "e9f51d15-a08d-5353-ae04-b508af25572c"
LIST_FEEDBACK_ISSUES_UUID        = "25e01f2e-2cad-5c33-aa5e-dd363d2074b7"
GET_ANTHROPIC_KEY_UUID           = "a8d55e58-1163-5277-94b7-1384125cec35"
REPORT_API_USAGE_UUID            = "864e82a2-cf13-5c03-a0f1-6c014ce48f65"
GET_API_USAGE_STATS_UUID         = "3d10ca59-8279-5c98-af7d-301bf469c6c1"
SESSION_HEARTBEAT_UUID           = "36d38928-a7b1-5b14-b3ba-edca5ba12027"
ASK_ADVISOR_UUID                 = "83ad6409-187e-5bb0-acc3-90b7f25c61d1"
ASK_TAX_RESEARCHER_UUID          = "5b16a8c1-59c9-554d-bcbe-4ade16eac755"


_DOMAIN_TOOLS = [
    ToolIdentity(tool_id=VERIFY_PASSPHRASE_UUID, capability="verify_passphrase", category="free", intent="Verify passphrase for timeout unlock"),
    ToolIdentity(tool_id=CREATE_SESSION_UUID, capability="create_session", category="free", intent="Create a tax session"),
    ToolIdentity(tool_id=GET_SESSION_UUID, capability="get_session", category="free", intent="Get session details"),
    ToolIdentity(tool_id=LIST_SESSIONS_UUID, capability="list_sessions", category="free", intent="List patron sessions"),
    ToolIdentity(tool_id=GET_RULES_UUID, capability="get_rules", category="free", intent="Get classification rules"),
    ToolIdentity(tool_id=COUNT_RULE_MATCHES_UUID, capability="count_rule_matches", category="free", intent="Count transactions matching a rule pattern"),
    ToolIdentity(tool_id=GET_IMPORT_STATS_UUID, capability="get_import_stats", category="free", intent="Get import statistics"),
    ToolIdentity(tool_id=LOAD_SHARE_TOKEN_UUID, capability="load_share_token", category="free", intent="Load a shared session"),
    ToolIdentity(tool_id=GET_TRANSACTIONS_UUID, capability="get_transactions", category="free", intent="Get transactions with filters"),
    ToolIdentity(tool_id=GET_TRANSACTIONS_PAGED_UUID, capability="get_transactions_paged", category="free", intent="Server-side filtered, grouped, sorted, paginated transactions"),
    ToolIdentity(tool_id=GET_SUMMARY_UUID, capability="get_summary", category="free", intent="Get grouped tax summary"),
    ToolIdentity(tool_id=IMPORT_CSV_UUID, capability="import_csv", category="free", intent="Import CSV transactions"),
    ToolIdentity(tool_id=SAVE_CLASSIFICATIONS_UUID, capability="save_classifications", category="free", intent="Bulk write classifications from FE"),
    ToolIdentity(tool_id=DELETE_CLASSIFICATION_UUID, capability="delete_classification", category="free", intent="Remove a classification (revert to unclassified)"),
    ToolIdentity(tool_id=RESET_CLASSIFICATIONS_UUID, capability="reset_classifications", category="free", intent="Delete all classifications, keeping transactions"),
    ToolIdentity(tool_id=DELETE_ACCOUNT_TRANSACTIONS_UUID, capability="delete_account_transactions", category="free", intent="Delete all transactions for a specific account"),
    ToolIdentity(tool_id=CLEAR_TRANSACTIONS_UUID, capability="clear_transactions", category="free", intent="Delete all transactions and classifications for a session"),
    ToolIdentity(tool_id=GET_AMOUNT_NEIGHBORS_UUID, capability="get_amount_neighbors", category="free", intent="Fetch transactions with same amount near a date"),
    ToolIdentity(tool_id=GET_ACCOUNTS_UUID, capability="get_accounts", category="free", intent="List accounts in session with their types"),
    ToolIdentity(tool_id=SET_ACCOUNT_TYPE_UUID, capability="set_account_type", category="free", intent="Set account type (bank, card, investment, loan)"),
    ToolIdentity(tool_id=SAVE_RULE_UUID, capability="save_rule", category="free", intent="Save a classification rule"),
    ToolIdentity(tool_id=DELETE_RULE_UUID, capability="delete_rule", category="free", intent="Delete a classification rule"),
    ToolIdentity(tool_id=APPLY_RULES_UUID, capability="apply_rules", category="free", intent="Apply rules to unclassified transactions"),
    ToolIdentity(tool_id=GET_CUSTOM_CATEGORIES_UUID, capability="get_custom_categories", category="free", intent="Get custom categories"),
    ToolIdentity(tool_id=SAVE_CUSTOM_CATEGORY_UUID, capability="save_custom_category", category="free", intent="Add a custom category/subcategory"),
    ToolIdentity(tool_id=DELETE_CUSTOM_CATEGORY_UUID, capability="delete_custom_category", category="free", intent="Delete a custom category"),
    ToolIdentity(tool_id=CREATE_SHARE_TOKEN_UUID, capability="create_share_token", category="free", intent="Create a session share token"),
    ToolIdentity(tool_id=REQUEST_UNLOCK_UUID, capability="request_unlock", category="free", intent="Request session unlock via Secure Courier"),
    ToolIdentity(tool_id=CHECK_UNLOCK_UUID, capability="check_unlock", category="free", intent="Check if session unlock was approved"),
    ToolIdentity(tool_id=GET_GITHUB_TOKEN_UUID, capability="get_github_token", category="free", intent="Get GitHub token for issue reporting"),
    ToolIdentity(tool_id=CREATE_FEEDBACK_ISSUE_UUID, capability="create_feedback_issue", category="free", intent="Create a GitHub issue for feedback"),
    ToolIdentity(tool_id=LIST_FEEDBACK_ISSUES_UUID, capability="list_feedback_issues", category="free", intent="List feedback issues for this patron"),
    ToolIdentity(tool_id=GET_ANTHROPIC_KEY_UUID, capability="get_anthropic_key", category="free", intent="Get Anthropic API key for FE classification"),
    ToolIdentity(tool_id=REPORT_API_USAGE_UUID, capability="report_api_usage", category="free", intent="Report Anthropic API usage from FE classification"),
    ToolIdentity(tool_id=GET_API_USAGE_STATS_UUID, capability="get_api_usage_stats", category="free", intent="Get aggregated API usage statistics"),
    ToolIdentity(tool_id=SESSION_HEARTBEAT_UUID, capability="session_heartbeat", category="free", intent="Presence heartbeat — who's active in this session"),
    ToolIdentity(tool_id=ASK_ADVISOR_UUID, capability="ask_advisor", category="free", intent="Ask the Financial Advisor about TaxSort"),
    ToolIdentity(tool_id=ASK_TAX_RESEARCHER_UUID, capability="ask_tax_researcher", category="free", intent="Ask the Tax Code Researcher about IRS provisions"),
]

TOOL_REGISTRY: dict[str, ToolIdentity] = {ti.tool_id: ti for ti in _DOMAIN_TOOLS}


async def _on_npub_proven(npub: str, payload: dict) -> None:
    """Store passphrase hash after npub ownership is proven."""
    from tools.verification import store_verification
    passphrase = payload.get("passphrase", "verified")
    await store_verification(npub, passphrase or "verified")


# ---------------------------------------------------------------------------
# OperatorRuntime
# ---------------------------------------------------------------------------

runtime = OperatorRuntime(
    tool_registry={**STANDARD_IDENTITIES, **TOOL_REGISTRY},
    operator_credential_template=CredentialTemplate(
        service="taxsort-operator",
        version=1,
        description="Operator credentials for BTCPay Lightning and Anthropic AI",
        fields={
            "btcpay_host": FieldSpec(
                required=True, sensitive=True,
                description="BTCPay Server URL (e.g. https://btcpay.example.com).",
            ),
            "btcpay_api_key": FieldSpec(
                required=True, sensitive=True,
                description="BTCPay Server API key.",
            ),
            "btcpay_store_id": FieldSpec(
                required=True, sensitive=True,
                description="BTCPay Store ID.",
            ),
            "anthropic_api_key": FieldSpec(
                required=True, sensitive=True,
                description="Anthropic API key for Claude AI classification.",
            ),
            "github_token": FieldSpec(
                required=False, sensitive=True,
                description="GitHub personal access token for creating feedback issues (optional).",
            ),
        },
    ),
    operator_credential_greeting=(
        "Hi \u2014 I'm TaxSort MCP, a Tollbooth service for personal tax "
        "transaction classification. To come online I need your "
        "BTCPay credentials and Anthropic API key."
    ),
    service_name="TaxSort MCP",
    credential_validator=_validate_taxsort_creds,
    npub_proof_field="passphrase",
    npub_proof_greeting=(
        "Hi \u2014 I'm TaxSort MCP. To verify you own this npub and "
        "protect your tax data, please reply with any passphrase. "
        "Your response will be encrypted and signed by your Nostr key."
    ),
    on_npub_proven=_on_npub_proven,
)

# ---------------------------------------------------------------------------
# Register standard DPYC tools from the wheel
# ---------------------------------------------------------------------------

tool = register_standard_tools(
    mcp,
    "taxsort",
    runtime,
    service_name="taxsort-mcp",
    service_version=__version__,
)

# ---------------------------------------------------------------------------
# Domain-specific MCP tools
# ---------------------------------------------------------------------------
# Domain schema is created lazily on first vault access (see db/neon.py).


# ── Verification ──────────────────────────────────────────────────────────
# npub ownership proof is handled by the wheel's standard
# request_npub_proof / receive_npub_proof tools. The on_npub_proven
# callback above stores the passphrase hash for domain use.

@tool
@runtime.paid_tool(capability_uuid("verify_passphrase"))
async def verify_passphrase(
    passphrase: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Verify a passphrase to unlock a timed-out session."""
    from tools.verification import verify_passphrase as _verify
    return await _verify(npub=npub, passphrase=passphrase)


# ── Sessions ──────────────────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("create_session"))
async def create_session(
    label: str = "",
    tax_year: int = 0,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Create a new TaxSort session for a tax year."""
    from tools.sessions import create_session as _create_session
    return await _create_session(owner_npub=npub, label=label, tax_year=tax_year)


@tool
@runtime.paid_tool(capability_uuid("get_session"))
async def get_session(
    session_id: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get session details and summary stats."""
    from tools.sessions import get_session as _get_session
    return await _get_session(session_id=session_id)


@tool
@runtime.paid_tool(capability_uuid("list_sessions"))
async def list_sessions(
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """List all sessions owned by the current patron."""
    from tools.sessions import list_sessions as _list_sessions
    return await _list_sessions(owner_npub=npub)


# ── Import ────────────────────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("import_csv"))
async def import_csv(
    session_id: str,
    content: str,
    filename: str,
    account_name: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Import a CSV file into a session. Content is the raw CSV text. Optional account_name overrides the filename-derived account."""
    from tools.imports import import_csv as _import_csv
    return await _import_csv(session_id=session_id, content=content, filename=filename, account_name=account_name)


@tool
@runtime.paid_tool(capability_uuid("get_import_stats"))
async def get_import_stats(
    session_id: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get import statistics for a session."""
    from tools.imports import get_import_stats as _get_import_stats
    return await _get_import_stats(session_id=session_id)


# ── Transactions ──────────────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("get_transactions"))
async def get_transactions(
    session_id: str,
    category: str = "",
    subcategory: str = "",
    month: str = "",
    search: str = "",
    account: str = "",
    date_from: str = "",
    date_to: str = "",
    unclassified_only: bool = False,
    limit: int = 200,
    offset: int = 0,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get transactions for a session with optional filters.

    Returns raw transactions LEFT JOINed with their classifications.
    Use unclassified_only=true to fetch pages of transactions needing
    classification by the FE. Use date_from/date_to for date range queries.
    """
    from tools.transactions import get_transactions as _get_transactions
    return await _get_transactions(
        session_id=session_id, category=category, subcategory=subcategory,
        month=month, search=search, account=account,
        date_from=date_from, date_to=date_to,
        unclassified_only=unclassified_only, limit=limit, offset=offset,
    )


@tool
@runtime.paid_tool(capability_uuid("get_transactions_paged"))
async def get_transactions_paged(
    session_id: str,
    category: str = "",
    subcategory: str = "",
    month: str = "",
    search: str = "",
    account: str = "",
    unclassified_only: bool = False,
    classified_only: bool = False,
    group_by: str = "none",
    group_sort: str = "asc",
    sort_col: str = "date",
    sort_dir: str = "asc",
    page: int = 0,
    page_size: int = 200,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Server-side filtered, grouped, sorted, paginated transactions.

    group_sort controls the order of groups (A-Z vs Z-A).
    sort_col + sort_dir control row order within each group.
    When group_by='none', only sort_col + sort_dir apply.
    """
    from tools.transactions import get_transactions_paged as _get_paged
    return await _get_paged(
        session_id=session_id, category=category, subcategory=subcategory,
        month=month, search=search, account=account,
        unclassified_only=unclassified_only, classified_only=classified_only,
        group_by=group_by, group_sort=group_sort,
        sort_col=sort_col, sort_dir=sort_dir,
        page=page, page_size=page_size,
    )


@tool
@runtime.paid_tool(capability_uuid("save_classifications"))
async def save_classifications(
    session_id: str,
    classifications: str = "[]",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Bulk write classifications from the FE.

    classifications is a JSON array of objects, each with:
      - id: raw_transaction_id
      - category, subcategory (required)
      - confidence, reason, merchant, description_override (optional)
      - classified_by: 'ai' | 'rule' | 'manual' (default 'ai')
    """
    import json as _json
    try:
        items = _json.loads(classifications) if isinstance(classifications, str) else classifications
    except _json.JSONDecodeError:
        return {"error": "Invalid JSON in classifications parameter"}

    from tools.transactions import save_classifications as _save
    return await _save(session_id=session_id, classifications=items)


@tool
@runtime.paid_tool(capability_uuid("delete_classification"))
async def delete_classification(
    session_id: str,
    transaction_id: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Remove a classification, reverting the transaction to unclassified."""
    from tools.transactions import delete_classification as _delete
    return await _delete(session_id=session_id, transaction_id=transaction_id)


@tool
@runtime.paid_tool(capability_uuid("clear_transactions"))
async def clear_transactions(
    session_id: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Delete all transactions and classifications for a session, so CSVs can be re-imported."""
    from tools.transactions import clear_transactions as _clear
    return await _clear(session_id=session_id)


@tool
@runtime.paid_tool(capability_uuid("delete_account_transactions"))
async def delete_account_transactions(
    session_id: str,
    account: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Delete all transactions and classifications for a specific imported account."""
    from tools.transactions import delete_account_transactions as _del
    return await _del(session_id=session_id, account=account)


@tool
@runtime.paid_tool(capability_uuid("reset_classifications"))
async def reset_classifications(
    session_id: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Delete all classifications but keep the imported transactions."""
    from tools.transactions import reset_classifications as _reset
    return await _reset(session_id=session_id)


@tool
@runtime.paid_tool(capability_uuid("get_amount_neighbors"))
async def get_amount_neighbors(
    session_id: str,
    amount: float,
    date: str,
    days: int = 14,
    exclude_id: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Fetch transactions with the same amount within ±days of a date. Used by the classifier to detect duplicates from overlapping CSV imports."""
    from tools.transactions import get_amount_neighbors as _get
    return await _get(session_id=session_id, amount=amount, date=date, days=days, exclude_id=exclude_id)


@tool
@runtime.paid_tool(capability_uuid("get_accounts"))
async def get_accounts(
    session_id: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """List all accounts in this session with their types and transaction counts."""
    from tools.accounts import get_accounts as _get
    return await _get(session_id=session_id)


@tool
@runtime.paid_tool(capability_uuid("set_account_type"))
async def set_account_type(
    session_id: str,
    account_name: str,
    account_type: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Set an account's type: bank (checking/savings), card (credit/debit), investment, or loan."""
    from tools.accounts import set_account_type as _set
    return await _set(session_id=session_id, account_name=account_name, account_type=account_type)



@tool
@runtime.paid_tool(capability_uuid("get_summary"))
async def get_summary(
    session_id: str,
    group_by: str = "taxline",
    scope: str = "tax",
    month: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get a grouped spending summary for tax reporting."""
    from tools.transactions import get_summary as _get_summary
    return await _get_summary(
        session_id=session_id, group_by=group_by, scope=scope, month=month,
    )


# ── Rules ─────────────────────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("get_rules"))
async def get_rules(
    session_id: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get all classification rules for the current patron."""
    from tools.rules import get_rules as _get_rules
    return await _get_rules(owner_npub=npub, session_id=session_id)


@tool
@runtime.paid_tool(capability_uuid("save_rule"))
async def save_rule(
    description_pattern: str,
    category: str,
    subcategory: str,
    new_description: str = "",
    amount_operator: str = "",
    amount_value: float | None = None,
    session_id: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Create a classification rule.

    Provide description_pattern (regex matched case-insensitively against
    the transaction description), category, and subcategory. Optionally add
    amount_operator (lt, lte, gt, gte, eq, neq) and amount_value to filter
    by amount. When the compound constraint matches, category, subcategory,
    and optionally description (new_description) are written.
    """
    from tools.rules import save_rule as _save_rule
    return await _save_rule(
        owner_npub=npub,
        description_pattern=description_pattern,
        category=category,
        subcategory=subcategory,
        new_description=new_description,
        amount_operator=amount_operator,
        amount_value=amount_value,
        session_id=session_id,
    )


@tool
@runtime.paid_tool(capability_uuid("delete_rule"))
async def delete_rule(
    rule_id: int,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Delete a classification rule by ID."""
    from tools.rules import delete_rule as _delete_rule
    return await _delete_rule(owner_npub=npub, rule_id=rule_id)


@tool
@runtime.paid_tool(capability_uuid("apply_rules"))
async def apply_rules(
    session_id: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Apply rules to unclassified transactions in a session."""
    from tools.rules import apply_rules as _apply_rules
    return await _apply_rules(owner_npub=npub, session_id=session_id)


@tool
@runtime.paid_tool(capability_uuid("count_rule_matches"))
async def count_rule_matches(
    session_id: str,
    description_pattern: str,
    amount_operator: str = "",
    amount_value: float | None = None,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Count how many transactions match a rule pattern (live preview)."""
    from tools.rules import count_rule_matches as _count
    return await _count(
        session_id=session_id,
        description_pattern=description_pattern,
        amount_operator=amount_operator,
        amount_value=amount_value,
    )


@tool
@runtime.paid_tool(capability_uuid("get_custom_categories"))
async def get_custom_categories(
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get custom categories defined by this user."""
    from tools.categories import get_custom_categories as _get
    return await _get(owner_npub=npub)


@tool
@runtime.paid_tool(capability_uuid("save_custom_category"))
async def save_custom_category(
    category: str,
    subcategory: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Add a custom category/subcategory (e.g. Personal / Auto Gas)."""
    from tools.categories import save_custom_category as _save
    return await _save(owner_npub=npub, category=category, subcategory=subcategory)


@tool
@runtime.paid_tool(capability_uuid("delete_custom_category"))
async def delete_custom_category(
    category_id: int,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Delete a custom category."""
    from tools.categories import delete_custom_category as _del
    return await _del(owner_npub=npub, category_id=category_id)


# ── Sharing ───────────────────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("create_share_token"))
async def create_share_token(
    session_id: str,
    expires_days: int = 30,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Create a share token so another user can access this session."""
    from tools.share import create_share_token as _create_share_token
    return await _create_share_token(
        owner_npub=npub, session_id=session_id,
        expires_days=expires_days,
    )


@tool
@runtime.paid_tool(capability_uuid("load_share_token"))
async def load_share_token(
    share_token: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Load a shared session via a share token."""
    from tools.share import load_share_token as _load_share_token
    return await _load_share_token(share_token=share_token)


# ── Feedback (GitHub token for frontend) ──────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("get_github_token"))
async def get_github_token(
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get the GitHub token for creating issues in the taxsort-mcp repo."""
    try:
        creds = await runtime.load_credentials(["github_token"])
        token = creds.get("github_token")
        if token:
            return {
                "token": token,
                "repo": "lonniev/taxsort-mcp",
                "scope": "issues",
            }
        return {"token": None, "message": "No GitHub token configured. Deliver one via Secure Courier."}
    except Exception as e:  # noqa: BLE001
        return {"token": None, "error": str(e)}


@tool
@runtime.paid_tool(capability_uuid("get_anthropic_key"))
async def get_anthropic_key(
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get the Anthropic API key for FE-driven classification."""
    try:
        creds = await runtime.load_credentials(["anthropic_api_key"])
        key = creds.get("anthropic_api_key")
        if key:
            return {"key": key}
        return {"key": None, "message": "No Anthropic API key configured. Deliver one via Secure Courier."}
    except Exception as e:  # noqa: BLE001
        return {"key": None, "error": str(e)}


# ── API Usage Reporting ──────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("report_api_usage"))
async def report_api_usage(
    session_id: str,
    calls: int,
    input_tokens: int,
    output_tokens: int,
    model: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Report Anthropic API usage from FE classification for cost tracking."""
    from tools.usage import report_usage
    return await report_usage(
        session_id=session_id, npub=npub,
        calls=calls, input_tokens=input_tokens,
        output_tokens=output_tokens, model=model,
    )


@tool
@runtime.paid_tool(capability_uuid("get_api_usage_stats"))
async def get_api_usage_stats(
    session_id: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Get aggregated API usage statistics for cost analysis."""
    from tools.usage import get_usage_stats
    return await get_usage_stats(session_id=session_id, npub=npub)


# ── Feedback ─────────────────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("create_feedback_issue"))
async def create_feedback_issue(
    title: str,
    body: str = "",
    category: str = "feedback",
    contact: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Create a GitHub issue for bug reports, feature requests, or feedback."""
    from tools.feedback import create_issue
    return await create_issue(npub=npub, title=title, body=body, category=category, contact=contact)


@tool
@runtime.paid_tool(capability_uuid("list_feedback_issues"))
async def list_feedback_issues(
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """List feedback issues submitted by this patron."""
    from tools.feedback import list_my_issues
    return await list_my_issues(npub=npub)


# ── Presence ──────────────────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("session_heartbeat"))
async def session_heartbeat(
    session_id: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Presence heartbeat. Returns who else is active in this session."""
    from tools.presence import heartbeat as _heartbeat
    return await _heartbeat(session_id=session_id, npub=npub)


# ── AI Advisors ───────────────────────────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("ask_advisor"))
async def ask_advisor(
    question: str,
    session_id: str = "",
    history: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Ask the Financial Advisor about using TaxSort."""
    import json as _json

    from tools.advisors import ask_advisor as _ask_advisor
    h = _json.loads(history) if history else []
    return await _ask_advisor(question=question, session_id=session_id, history=h)


@tool
@runtime.paid_tool(capability_uuid("ask_tax_researcher"))
async def ask_tax_researcher(
    question: str,
    session_id: str = "",
    history: str = "",
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Ask the Tax Code Researcher about IRS provisions."""
    import json as _json

    from tools.advisors import ask_tax_researcher as _ask_tax_researcher
    h = _json.loads(history) if history else []
    return await _ask_tax_researcher(question=question, session_id=session_id, history=h)


# ── Session Unlock (Nostr DM exchange) ─────────────────────────────────────

@tool
@runtime.paid_tool(capability_uuid("request_unlock"))
async def request_unlock(
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Request a session unlock after timeout."""
    from tools.session_lock import request_unlock as _request_unlock

    dm_sent = False
    dm_error = None

    try:
        courier = await runtime.courier()
        await courier.open_channel(
            service="taxsort-patron",
            greeting=(
                "Your TaxSort session has timed out. "
                "Reply with the exact words: Approve Unlock"
            ),
            recipient_npub=npub,
        )
        dm_sent = True
    except Exception as e:  # noqa: BLE001
        dm_error = str(e)
        logger.warning("Failed to send unlock DM to %s: %s", npub[:20], e)

    result = await _request_unlock(npub)
    result["dm_sent"] = dm_sent
    if dm_error:
        result["dm_error"] = dm_error
    return result


@tool
@runtime.paid_tool(capability_uuid("check_unlock"))
async def check_unlock(
    response: str,
    npub: NpubField = "", dpop_token: str = "",
) -> dict[str, Any]:
    """Check if the unlock response is valid."""
    from tools.session_lock import check_unlock as _check_unlock
    return await _check_unlock(npub=npub, response=response)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------


def main() -> None:
    """Main entry point for the server."""
    from tollbooth import validate_operator_tools

    missing = validate_operator_tools(mcp, "taxsort")
    if missing:
        import sys
        print(
            f"\u26a0 Missing base-catalog tools: {', '.join(missing)}",
            file=sys.stderr,
        )
    mcp.run()


if __name__ == "__main__":
    main()