fin/cli/fin_mcp/server.py

"""MCP server wrapping FinClient.

Entry point: `fin-mcp`.

Transport is selected with FIN_MCP_TRANSPORT:
  stdio           (default) — for local clients that spawn the process
  streamable-http           — for running as a network service; binds
                              FIN_MCP_HOST:FIN_MCP_PORT and serves at /mcp

Default mode: read-only (15 tools).
Set FIN_MCP_ALLOW_WRITES=1 to additionally enable 24 write tools.
"""

from __future__ import annotations

import json
import os
from typing import Any, Literal

try:
    from mcp.server.fastmcp import FastMCP
    from mcp.server.transport_security import TransportSecuritySettings
    from mcp.types import ToolAnnotations
except ImportError as exc:
    raise ImportError(
        "fin-mcp requires the 'mcp' extra: pip install 'fin-cli[mcp]'"
    ) from exc

from fin_cli.client import FinClient

TRUTHY = {"1", "true", "yes", "on"}


class _LazyClient:
    """Defers FinClient construction until the first tool call.

    FinClient raises when FIN_API_URL is unset. Doing that at import/startup
    kills the process before the MCP handshake, so the client only ever sees
    "server failed to start" with no cause. Deferring it turns the same
    misconfiguration into a readable error on the first tool result.
    """

    def __init__(self) -> None:
        self._client: FinClient | None = None

    def __getattr__(self, name: str) -> Any:
        if self._client is None:
            self._client = FinClient()
        return getattr(self._client, name)


def _match_by_name(rows: list[dict[str, Any]] | None, name: str) -> list[dict[str, Any]]:
    """Resolve a human name to rows, preferring exact matches.

    An exact case-insensitive hit wins outright, so searching "Food" cannot be
    drowned out by "Fast Food". Only when nothing matches exactly do we fall
    back to substring matching and let the caller disambiguate.
    """
    needle = name.strip().lower()
    rows = rows or []
    exact = [r for r in rows if (r.get("name") or "").strip().lower() == needle]
    if exact:
        return exact
    return [r for r in rows if needle in (r.get("name") or "").lower()]


def _instructions(allow_writes: bool) -> str:
    mode_line = (
        "writes ENABLED"
        if allow_writes
        else (
            "READ-ONLY — write tools are disabled; "
            "set FIN_MCP_ALLOW_WRITES=1 on the server to enable"
        )
    )
    base = (
        "fin is a single-user, self-hosted personal finance app tracking bank accounts, "
        "investments, RSU grants, Splitwise balances, receivables, and net worth in EUR.\n\n"
        "Tool return convention: every tool returns {\"result\": <payload>} where <payload> "
        "is the raw API JSON (list, dict, or null). Unwrap result before displaying.\n\n"
        f"Current mode: {mode_line}.\n\n"
        "Sign convention: amounts are negative for money out (expenses, transfers from an "
        "account), positive for money in (income, transfers to an account)."
    )
    if not allow_writes:
        return base
    return base + (
        "\n\nWrite discipline (applies to every non-read tool):\n"
        "- Confirm amounts, accounts and dates with the user before writing.\n"
        "- Pass a unique client_ref on create_transaction, create_transfer and "
        "create_settlement. If the call fails ambiguously, retry with the SAME "
        "client_ref — the write is returned rather than repeated, so you cannot "
        "double-book. Tools without a client_ref parameter are not yet protected: "
        "there, list recent transactions and check for an identical row first.\n"
        "- Accounts and categories are addressed by integer id, never by name. "
        "Resolve the id with find_account / find_category first; if the lookup "
        "returns nothing or more than one row, ask the user instead of guessing."
    )


def create_server(
    client: FinClient | None = None,
    allow_writes: bool | None = None,
) -> FastMCP:
    """Create the fin MCP server.

    Args:
        client: FinClient to use; if None, built from FIN_API_URL / FIN_API_TOKEN env vars.
        allow_writes: if None, read from FIN_MCP_ALLOW_WRITES env var.
    """
    if allow_writes is None:
        allow_writes = os.getenv("FIN_MCP_ALLOW_WRITES", "").lower() in TRUTHY
    if client is None:
        # Built lazily on first tool call. Constructing here would abort the
        # process before the MCP handshake when FIN_API_URL is unset, leaving
        # the calling client with nothing but "server failed to start".
        client = _LazyClient()

    mcp = FastMCP("fin", instructions=_instructions(allow_writes))

    # ------------------------------------------------------------------
    # Read tools (always registered)
    # ------------------------------------------------------------------

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def health() -> dict[str, Any]:
        """Check that the fin API is reachable and healthy."""
        return {"result": client.health()}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def get_net_worth() -> dict[str, Any]:
        """Return the current net worth breakdown in EUR, grouped by account."""
        return {"result": client.get_net_worth()}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def get_net_worth_history(
        range: Literal["1M", "3M", "6M", "1Y", "all"] = "all"
    ) -> dict[str, Any]:
        """Return net worth snapshots ordered by date ascending.

        Args:
            range: Time range — one of: 1M, 3M, 6M, 1Y, all (default: all).
        """
        return {"result": client.get_net_worth_history(range=range)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def list_accounts() -> dict[str, Any]:
        """List all active accounts (bank, investment, RSU, Splitwise, receivable, payable)."""
        return {"result": client.list_accounts()}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def list_categories() -> dict[str, Any]:
        """List all spending and income categories."""
        return {"result": client.list_categories()}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def find_account(name: str) -> dict[str, Any]:
        """Resolve an account name to the integer id that other tools require.

        Every other tool addresses accounts by integer id, so use this instead
        of listing all accounts and matching names by hand.

        An exact case-insensitive match wins outright; otherwise all substring
        matches are returned. An empty list means no such account — ask the
        user rather than guessing an id. More than one result means the name
        was ambiguous, so ask which one they meant.

        Args:
            name: Full or partial account name, e.g. "checking".
        """
        return {"result": _match_by_name(client.list_accounts(), name)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def find_category(name: str) -> dict[str, Any]:
        """Resolve a category name to the integer id that other tools require.

        Same matching rules as find_account: exact match wins, otherwise
        substring matches. Empty means no such category — do not invent an id,
        and do not silently create one.

        Args:
            name: Full or partial category name, e.g. "groceries".
        """
        return {"result": _match_by_name(client.list_categories(), name)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def list_transactions(
        account_id: int | None = None,
        category_id: int | None = None,
        month: str | None = None,
        limit: int | None = None,
    ) -> dict[str, Any]:
        """List transactions, optionally filtered.

        Args:
            account_id: Filter to one account (integer id).
            category_id: Filter to one category (integer id).
            month: Filter to one calendar month in YYYY-MM format.
            limit: Maximum number of rows to return (most recent first).
        """
        return {
            "result": client.list_transactions(
                account_id=account_id,
                category_id=category_id,
                month=month,
                limit=limit,
            )
        }

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def get_holdings(account_id: int) -> dict[str, Any]:
        """Return holdings with live P&L for one investment or RSU account.

        Args:
            account_id: Integer id of an investment or RSU account.
        """
        return {"result": client.get_holdings(account_id=account_id)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def list_stock_transactions(account_id: int | None = None) -> dict[str, Any]:
        """List stock transactions (buy/sell/rsu_vest), optionally filtered by account.

        Args:
            account_id: If provided, return only trades for this account.
        """
        return {"result": client.list_stock_transactions(account_id=account_id)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def list_rsu_grants(include_vested: bool = False) -> dict[str, Any]:
        """List RSU grants.

        Args:
            include_vested: If True, include already-vested grants (default: False).
        """
        return {"result": client.list_rsu_grants(include_vested=include_vested)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def preview_rsu_schedule(
        account_id: int,
        ticker: str,
        grant_date: str,
        total_gross_shares: float,
        cliff_months: int = 12,
        total_months: int = 36,
        frequency_months: int = 3,
        withholding_pct: float = 0.5,
        notes: str | None = None,
    ) -> dict[str, Any]:
        """Preview the vesting schedule for a hypothetical RSU grant without writing anything.

        Pure computation — safe in read-only mode; no data is stored.

        Args:
            account_id: RSU account to associate with.
            ticker: Stock ticker symbol (1-12 chars, alphanumeric/dot/dash).
            grant_date: Grant date in YYYY-MM-DD format.
            total_gross_shares: Total gross shares in the grant (>0).
            cliff_months: Months until first vest (1-60, default 12).
            total_months: Total vesting period in months (12-120, default 36).
            frequency_months: Vesting frequency in months (1-12, default 3).
            withholding_pct: Tax withholding fraction (0-1, default 0.5).
            notes: Optional notes (max 500 chars).
        """
        data: dict[str, Any] = {
            "account_id": account_id,
            "ticker": ticker,
            "grant_date": grant_date,
            "total_gross_shares": total_gross_shares,
            "cliff_months": cliff_months,
            "total_months": total_months,
            "frequency_months": frequency_months,
            "withholding_pct": withholding_pct,
        }
        if notes is not None:
            data["notes"] = notes
        return {"result": client.preview_rsu_schedule(data=data)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def get_spending_month(month: str, account_id: int | None = None) -> dict[str, Any]:
        """Return spending breakdown by category for one calendar month.

        Args:
            month: Month in YYYY-MM format (required).
            account_id: If provided, restrict to one account.
        """
        return {"result": client.get_spending_month(month=month, account_id=account_id)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def get_spending_summary(
        mode: Literal["yearly", "custom"],
        year: int | None = None,
        start: str | None = None,
        end: str | None = None,
    ) -> dict[str, Any]:
        """Aggregate spending over a time range.

        Args:
            mode: Aggregation mode — "yearly" (requires year) or "custom" (requires start + end).
            year: Calendar year (e.g. 2026); used with mode=yearly.
            start: Start date in YYYY-MM-DD format; used with mode=custom.
            end: End date in YYYY-MM-DD format; used with mode=custom.
        """
        params: dict[str, Any] = {"mode": mode}
        if year is not None:
            params["year"] = year
        if start is not None:
            params["start"] = start
        if end is not None:
            params["end"] = end
        return {"result": client.get_spending_summary(**params)}

    @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
    def get_fx_rate(
        from_currency: str = "EUR",
        to_currency: str = "USD",
    ) -> dict[str, Any]:
        """Return the cached FX rate between two currencies.

        Args:
            from_currency: Source currency code (default: EUR).
            to_currency: Target currency code (default: USD).
        """
        return {
            "result": client.get_fx_rate(
                from_currency=from_currency, to_currency=to_currency
            )
        }

    # ------------------------------------------------------------------
    # Write tools (registered only when allow_writes is True)
    # ------------------------------------------------------------------

    if allow_writes:

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_account(
            name: str,
            type: Literal["bank", "investment", "rsu", "splitwise", "receivable", "payable"],
            currency: str = "EUR",
            notes: str | None = None,
            manual_override_balance: float | None = None,
            is_active: bool = True,
            initial_balance: float | None = None,
        ) -> dict[str, Any]:
            """Create a new account.

            Args:
                name: Account name (max 100 chars).
                type: One of: bank, investment, rsu, splitwise, receivable, payable.
                currency: Currency code (default EUR).
                notes: Optional notes.
                manual_override_balance: If set, overrides the computed balance display.
                is_active: Whether the account is active (default True).
                initial_balance: If set, creates an opening-balance income transaction.
            """
            data: dict[str, Any] = {"name": name, "type": type, "currency": currency, "is_active": is_active}
            if notes is not None:
                data["notes"] = notes
            if manual_override_balance is not None:
                data["manual_override_balance"] = manual_override_balance
            if initial_balance is not None:
                data["initial_balance"] = initial_balance
            return {"result": client.create_account(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def update_account(
            account_id: int,
            name: str | None = None,
            type: Literal["bank", "investment", "rsu", "splitwise", "receivable", "payable"] | None = None,
            currency: str | None = None,
            notes: str | None = None,
            manual_override_balance: float | None = None,
            is_active: bool | None = None,
            clear_manual_override_balance: bool = False,
        ) -> dict[str, Any]:
            """Update fields on an existing account. Omitted fields are left unchanged.

            Args:
                account_id: Integer id of the account to update.
                name: New account name (max 100 chars).
                type: New account type.
                currency: New currency code.
                notes: New notes text.
                manual_override_balance: New override for the displayed balance.
                is_active: Set False to deactivate, True to reactivate.
                clear_manual_override_balance: Set True to remove an existing
                    override so the balance is computed from transactions again.
            """
            data: dict[str, Any] = {}
            if name is not None:
                data["name"] = name
            if type is not None:
                data["type"] = type
            if currency is not None:
                data["currency"] = currency
            if notes is not None:
                data["notes"] = notes
            if manual_override_balance is not None:
                data["manual_override_balance"] = manual_override_balance
            if is_active is not None:
                data["is_active"] = is_active
            if clear_manual_override_balance:
                data["manual_override_balance"] = None
            return {"result": client.update_account(account_id=account_id, data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(
                readOnlyHint=False, destructiveHint=False, idempotentHint=True
            )
        )
        def deactivate_account(account_id: int) -> dict[str, Any]:
            """Deactivate (soft-delete) an account. Reversible via update_account with is_active=True.

            Args:
                account_id: Integer id of the account to deactivate.
            """
            return {"result": client.deactivate_account(account_id=account_id)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def adjust_account_balance(account_id: int, target_balance: float) -> dict[str, Any]:
            """Adjust a bank account's balance to a target value by creating a correction transaction.

            Args:
                account_id: Integer id of a bank account.
                target_balance: The desired balance in the account's currency.
            """
            return {
                "result": client.adjust_account_balance(
                    account_id=account_id, target_balance=target_balance
                )
            }

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_category(
            name: str,
            color: str,
            icon: str,
            type: Literal["expense", "income"] = "expense",
        ) -> dict[str, Any]:
            """Create a new spending or income category.

            Args:
                name: Category name (max 100 chars).
                color: CSS color string (max 20 chars), e.g. "#e74c3c".
                icon: Emoji or icon name (max 20 chars).
                type: Either "expense" (default) or "income".
            """
            data: dict[str, Any] = {"name": name, "color": color, "icon": icon, "type": type}
            return {"result": client.create_category(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def update_category(
            category_id: int,
            name: str | None = None,
            color: str | None = None,
            icon: str | None = None,
            type: Literal["expense", "income"] | None = None,
        ) -> dict[str, Any]:
            """Update fields on an existing category. Omitted fields are left unchanged.

            Args:
                category_id: Integer id of the category to update.
                name: New category name (max 100 chars).
                color: New CSS color string (max 20 chars), e.g. "#e74c3c".
                icon: New emoji or icon name (max 20 chars).
                type: New category type.
            """
            data: dict[str, Any] = {}
            if name is not None:
                data["name"] = name
            if color is not None:
                data["color"] = color
            if icon is not None:
                data["icon"] = icon
            if type is not None:
                data["type"] = type
            return {"result": client.update_category(category_id=category_id, data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True)
        )
        def delete_category(category_id: int) -> dict[str, Any]:
            """Delete a category. Transactions referencing it will have their category set to null.

            Args:
                category_id: Integer id of the category to delete.
            """
            return {"result": client.delete_category(category_id=category_id)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_transaction(
            date: str,
            account_id: int,
            amount: float,
            type: Literal["expense", "income"],
            currency: str = "EUR",
            category_id: int | None = None,
            description: str | None = None,
            fee: float = 0,
            exchange_rate_used: float | None = None,
            client_ref: str | None = None,
        ) -> dict[str, Any]:
            """Create a single expense or income transaction.

            Handles standalone expenses and income only. Transfers, settlements and
            Splitwise entries are two-leg writes that MUST go through their own tools —
            create_transfer, create_settlement, create_split_expense or
            create_splitwise_i_paid. Writing one leg here would leave the ledger
            unbalanced, so this tool cannot express those types.

            Args:
                date: Transaction date in YYYY-MM-DD format.
                account_id: Integer id of the bank account.
                amount: Amount in the account's currency. Negative for expenses, positive for income.
                type: "expense" or "income" only.
                currency: Currency code (default EUR).
                category_id: Optional integer category id.
                description: Optional description (max 500 chars).
                fee: Fee amount (included in balance calculation; default 0).
                exchange_rate_used: Optional FX rate used for this transaction.
                client_ref: Optional idempotency key. If a write with this key already
                    exists it is returned unchanged instead of booking a second time.
                    Set one before any write you might have to retry.
            """
            data: dict[str, Any] = {
                "date": date,
                "account_id": account_id,
                "amount": amount,
                "type": type,
                "currency": currency,
                "fee": fee,
            }
            if category_id is not None:
                data["category_id"] = category_id
            if description is not None:
                data["description"] = description
            if exchange_rate_used is not None:
                data["exchange_rate_used"] = exchange_rate_used
            if client_ref is not None:
                data["client_ref"] = client_ref
            return {"result": client.create_transaction(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def update_transaction(
            transaction_id: int,
            amount: float | None = None,
            description: str | None = None,
            category_id: int | None = None,
            date: str | None = None,
            clear_category: bool = False,
        ) -> dict[str, Any]:
            """Update an existing expense or income transaction. Omitted fields are unchanged.

            Note: the API rejects updates to linked transfer/settlement legs with 422.
            To change a transfer, delete the whole group and re-enter.

            Args:
                transaction_id: Integer id of the transaction to update.
                amount: New amount (negative for expenses, positive for income).
                description: New description (max 500 chars).
                category_id: New category integer id.
                date: New date in YYYY-MM-DD format.
                clear_category: Set True to remove the category entirely.
            """
            data: dict[str, Any] = {}
            if amount is not None:
                data["amount"] = amount
            if description is not None:
                data["description"] = description
            if category_id is not None:
                data["category_id"] = category_id
            if date is not None:
                data["date"] = date
            if clear_category:
                data["category_id"] = None
            return {
                "result": client.update_transaction(
                    transaction_id=transaction_id, data=data
                )
            }

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True)
        )
        def delete_transaction(transaction_id: int) -> dict[str, Any]:
            """Delete a transaction. If the transaction is part of a transfer/settlement group,
            the entire linked group is deleted atomically.

            Args:
                transaction_id: Integer id of the transaction (or any leg of a linked group).
            """
            return {"result": client.delete_transaction(transaction_id=transaction_id)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_transfer(
            from_account_id: int,
            to_account_id: int,
            amount: float,
            date: str,
            currency: str = "EUR",
            fee: float = 0,
            description: str = "Transfer",
            to_amount: float | None = None,
            to_currency: str | None = None,
            client_ref: str | None = None,
        ) -> dict[str, Any]:
            """Create an atomic two-leg transfer between two accounts.

            For cross-currency transfers, provide to_amount and to_currency.

            Args:
                from_account_id: Source account integer id.
                to_account_id: Destination account integer id.
                amount: Amount in the source currency (positive).
                date: Transfer date in YYYY-MM-DD format.
                currency: Source currency code (default EUR).
                fee: Fee deducted from source account (default 0).
                description: Description for both legs (default "Transfer").
                to_amount: Amount received in destination currency (for FX transfers).
                to_currency: Destination currency code (for FX transfers).
                client_ref: Optional idempotency key. If a write with this key already
                    exists it is returned unchanged instead of booking a second time.
                    Set one before any write you might have to retry.
            """
            data: dict[str, Any] = {
                "from_account_id": from_account_id,
                "to_account_id": to_account_id,
                "amount": amount,
                "currency": currency,
                "fee": fee,
                "date": date,
                "description": description,
            }
            if to_amount is not None:
                data["to_amount"] = to_amount
            if to_currency is not None:
                data["to_currency"] = to_currency
            if client_ref is not None:
                data["client_ref"] = client_ref
            return {"result": client.create_transfer(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_settlement(
            bank_account_id: int,
            splitwise_account_id: int,
            amount: float,
            date: str,
            direction: Literal["bank_to_splitwise", "splitwise_to_bank"],
            currency: str = "EUR",
            client_ref: str | None = None,
        ) -> dict[str, Any]:
            """Create an atomic Splitwise settlement between a bank account and a Splitwise account.

            Args:
                bank_account_id: Integer id of the bank account.
                splitwise_account_id: Integer id of the Splitwise account.
                amount: Settlement amount (positive).
                date: Settlement date in YYYY-MM-DD format.
                direction: "bank_to_splitwise" (you paid the Splitwise balance)
                           or "splitwise_to_bank" (Splitwise partner paid you).
                currency: Currency code (default EUR).
                client_ref: Optional idempotency key. If a write with this key already
                    exists it is returned unchanged instead of booking a second time.
                    Set one before any write you might have to retry.
            """
            data: dict[str, Any] = {
                "bank_account_id": bank_account_id,
                "splitwise_account_id": splitwise_account_id,
                "amount": amount,
                "currency": currency,
                "date": date,
                "direction": direction,
            }
            if client_ref is not None:
                data["client_ref"] = client_ref
            return {"result": client.create_settlement(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_split_expense(
            bank_account_id: int,
            receivable_account_id: int,
            total_amount: float,
            partner_share: float,
            date: str,
            currency: str = "EUR",
            category_id: int | None = None,
            description: str | None = None,
        ) -> dict[str, Any]:
            """Create a shared bill where you paid and a partner owes you their share.

            Records the full expense on the bank account and a receivable for the partner's share.

            Args:
                bank_account_id: Integer id of the bank account you paid from.
                receivable_account_id: Integer id of the receivable account for the partner.
                total_amount: Total bill amount (>0).
                partner_share: Partner's share of the bill (>0, must be less than total_amount).
                date: Date in YYYY-MM-DD format.
                currency: Currency code (default EUR).
                category_id: Optional integer category id.
                description: Optional description (max 500 chars).
            """
            data: dict[str, Any] = {
                "bank_account_id": bank_account_id,
                "receivable_account_id": receivable_account_id,
                "total_amount": total_amount,
                "partner_share": partner_share,
                "currency": currency,
                "date": date,
            }
            if category_id is not None:
                data["category_id"] = category_id
            if description is not None:
                data["description"] = description
            return {"result": client.create_split_expense(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_splitwise_i_paid(
            bank_account_id: int,
            splitwise_account_id: int,
            total_amount: float,
            partner_share: float,
            date: str,
            currency: str = "EUR",
            category_id: int | None = None,
            description: str = "Splitwise: I paid",
        ) -> dict[str, Any]:
            """Record a Splitwise expense where you paid the full bill and the group owes you.

            Args:
                bank_account_id: Integer id of the bank account you paid from.
                splitwise_account_id: Integer id of the Splitwise account.
                total_amount: Total bill amount (>0).
                partner_share: Partner's share (>0, must be less than total_amount).
                date: Date in YYYY-MM-DD format.
                currency: Currency code (default EUR).
                category_id: Optional integer category id.
                description: Description (default "Splitwise: I paid").
            """
            data: dict[str, Any] = {
                "bank_account_id": bank_account_id,
                "splitwise_account_id": splitwise_account_id,
                "total_amount": total_amount,
                "partner_share": partner_share,
                "currency": currency,
                "date": date,
                "description": description,
            }
            if category_id is not None:
                data["category_id"] = category_id
            return {"result": client.create_splitwise_i_paid(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def reconcile(
            bank_account_id: int,
            other_account_id: int,
            amount: float,
            date: str,
            description: str | None = None,
        ) -> dict[str, Any]:
            """Settle a receivable or payable against a bank account.

            For receivables (someone pays you back): bank receives +amount, other account −amount.
            For payables (you pay someone back): bank sends −amount, other account +amount.
            Both legs are typed 'settlement' and do not count as income or spending.

            Args:
                bank_account_id: Integer id of the bank account.
                other_account_id: Integer id of the payable or receivable account.
                amount: Amount to settle (>0).
                date: Settlement date in YYYY-MM-DD format.
                description: Optional description.
            """
            data: dict[str, Any] = {
                "bank_account_id": bank_account_id,
                "other_account_id": other_account_id,
                "amount": amount,
                "date": date,
            }
            if description is not None:
                data["description"] = description
            return {"result": client.reconcile(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_stock_transaction(
            account_id: int,
            ticker: str,
            type: Literal["buy", "sell", "rsu_vest"],
            shares: float,
            price_per_share: float,
            total_cost: float,
            date: str,
            fee: float = 0,
            currency: str = "USD",
            exchange_rate_used: float | None = None,
            notes: str | None = None,
        ) -> dict[str, Any]:
            """Record a stock buy, sell, or RSU vest. Updates the holding immediately.

            For a buy funded by a bank account use funded_buy instead (records
            both the trade and the cash withdrawal atomically).

            Args:
                account_id: Integer id of an investment or RSU account.
                ticker: Ticker symbol (1-12 chars, alphanumeric/dot/dash).
                type: "buy", "sell", or "rsu_vest".
                shares: Number of shares (>0).
                price_per_share: Price per share (>0).
                total_cost: Total cost including fees (>=0).
                date: Trade date in YYYY-MM-DD format.
                fee: Commission/fee (default 0).
                currency: Currency of the price (default USD).
                exchange_rate_used: Optional FX rate used.
                notes: Optional notes (max 500 chars).
            """
            data: dict[str, Any] = {
                "account_id": account_id,
                "ticker": ticker,
                "type": type,
                "shares": shares,
                "price_per_share": price_per_share,
                "fee": fee,
                "total_cost": total_cost,
                "date": date,
                "currency": currency,
            }
            if exchange_rate_used is not None:
                data["exchange_rate_used"] = exchange_rate_used
            if notes is not None:
                data["notes"] = notes
            return {"result": client.create_stock_transaction(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def funded_buy(
            account_id: int,
            ticker: str,
            shares: float,
            price_per_share: float,
            total_cost: float,
            funded_by_account_id: int,
            date: str,
            type: Literal["buy", "sell"] = "buy",
            fee: float = 0,
            currency: str = "USD",
            exchange_rate_used: float | None = None,
            notes: str | None = None,
        ) -> dict[str, Any]:
            """Buy stock and simultaneously deduct the cost from a bank account (atomic two-leg write).

            Args:
                account_id: Integer id of the investment account receiving the shares.
                ticker: Ticker symbol (1-12 chars, alphanumeric/dot/dash).
                shares: Number of shares purchased (>0).
                price_per_share: Price per share (>0).
                total_cost: Total cost including fees (>=0).
                funded_by_account_id: Integer id of the bank account to debit.
                date: Trade date in YYYY-MM-DD format.
                type: "buy" or "sell" (default "buy").
                fee: Commission/fee (default 0).
                currency: Currency of the price (default USD).
                exchange_rate_used: Optional FX rate used.
                notes: Optional notes (max 500 chars).
            """
            data: dict[str, Any] = {
                "account_id": account_id,
                "ticker": ticker,
                "type": type,
                "shares": shares,
                "price_per_share": price_per_share,
                "fee": fee,
                "total_cost": total_cost,
                "date": date,
                "currency": currency,
                "funded_by_account_id": funded_by_account_id,
            }
            if exchange_rate_used is not None:
                data["exchange_rate_used"] = exchange_rate_used
            if notes is not None:
                data["notes"] = notes
            return {"result": client.funded_buy(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True)
        )
        def delete_stock_transaction(stock_txn_id: int) -> dict[str, Any]:
            """Delete a stock transaction. Holdings are recomputed server-side after deletion.

            Args:
                stock_txn_id: Integer id of the stock transaction to delete.
            """
            return {"result": client.delete_stock_transaction(stock_txn_id=stock_txn_id)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def create_rsu_grant(
            account_id: int,
            ticker: str,
            vest_date: str,
            gross_shares: float,
            withholding_pct: float = 0.5,
            notes: str | None = None,
        ) -> dict[str, Any]:
            """Create an RSU grant record (does not vest shares — use vest_rsu_grant for that).

            Args:
                account_id: Integer id of an RSU account.
                ticker: Ticker symbol (1-12 chars, alphanumeric/dot/dash).
                vest_date: Planned vesting date in YYYY-MM-DD format.
                gross_shares: Total gross shares in the grant (>0).
                withholding_pct: Tax withholding fraction (0-1, default 0.5).
                notes: Optional notes (max 500 chars).
            """
            data: dict[str, Any] = {
                "account_id": account_id,
                "ticker": ticker,
                "vest_date": vest_date,
                "gross_shares": gross_shares,
                "withholding_pct": withholding_pct,
            }
            if notes is not None:
                data["notes"] = notes
            return {"result": client.create_rsu_grant(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=True)
        )
        def delete_rsu_grant(grant_id: int) -> dict[str, Any]:
            """Delete an RSU grant. Already-vested grants are refused (API returns 422).

            Args:
                grant_id: Integer id of the RSU grant to delete.
            """
            return {"result": client.delete_rsu_grant(grant_id=grant_id)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def vest_rsu_grant(
            grant_id: int,
            withholding_pct_override: float | None = None,
            net_shares: int | None = None,
        ) -> dict[str, Any]:
            """Vest an RSU grant, creating a stock transaction for the net shares received.

            Args:
                grant_id: Integer id of the RSU grant to vest.
                withholding_pct_override: Override the grant's withholding percentage (0-1).
                net_shares: Override the computed net shares (integer >=1).
            """
            data: dict[str, Any] = {}
            if withholding_pct_override is not None:
                data["withholding_pct_override"] = withholding_pct_override
            if net_shares is not None:
                data["net_shares"] = net_shares
            return {"result": client.vest_rsu_grant(grant_id=grant_id, data=data or None)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def generate_rsu_schedule(
            account_id: int,
            ticker: str,
            grant_date: str,
            total_gross_shares: float,
            cliff_months: int = 12,
            total_months: int = 36,
            frequency_months: int = 3,
            withholding_pct: float = 0.5,
            notes: str | None = None,
        ) -> dict[str, Any]:
            """Generate and persist the full RSU vesting schedule (creates grant records).

            Use preview_rsu_schedule to see the schedule without saving anything.

            Args:
                account_id: RSU account integer id.
                ticker: Ticker symbol (1-12 chars, alphanumeric/dot/dash).
                grant_date: Grant date in YYYY-MM-DD format.
                total_gross_shares: Total gross shares in the grant (>0).
                cliff_months: Months until first vest (1-60, default 12).
                total_months: Total vesting period in months (12-120, default 36).
                frequency_months: Vesting frequency in months (1-12, default 3).
                withholding_pct: Tax withholding fraction (0-1, default 0.5).
                notes: Optional notes (max 500 chars).
            """
            data: dict[str, Any] = {
                "account_id": account_id,
                "ticker": ticker,
                "grant_date": grant_date,
                "total_gross_shares": total_gross_shares,
                "cliff_months": cliff_months,
                "total_months": total_months,
                "frequency_months": frequency_months,
                "withholding_pct": withholding_pct,
            }
            if notes is not None:
                data["notes"] = notes
            return {"result": client.generate_rsu_schedule(data=data)}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def refresh_prices() -> dict[str, Any]:
            """Trigger a background refresh of all stock prices and FX rates.

            Returns 202 immediately; the refresh runs asynchronously.
            """
            return {"result": client.refresh_prices()}

        @mcp.tool(
            annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False)
        )
        def fetch_ticker_price(ticker: str) -> dict[str, Any]:
            """Fetch and cache the current price for a single ticker symbol from yfinance.

            WARNING: the upstream fetch is synchronous and retries internally.
            Unknown tickers routinely exceed the client's 30s timeout and come
            back as a request error rather than a result. Do not call in a loop.

            Args:
                ticker: Ticker symbol to fetch (e.g. "AAPL", "MSFT").
            """
            return {"result": client.fetch_ticker_price(ticker=ticker)}

    # ------------------------------------------------------------------
    # Resources — stable reference data every task needs. Exposing these
    # alongside the list_* tools lets a client pull them into context once
    # instead of spending a tool call per turn re-listing them.
    # ------------------------------------------------------------------

    @mcp.resource(
        "fin://accounts",
        name="accounts",
        description=(
            "All active fin accounts, with the integer id, type, currency and "
            "balance of each. Look up account ids here before writing."
        ),
        mime_type="application/json",
    )
    def accounts_resource() -> str:
        return json.dumps(client.list_accounts(), indent=2, default=str)

    @mcp.resource(
        "fin://categories",
        name="categories",
        description=(
            "All spending and income categories, with the integer id of each. "
            "Look up category ids here before writing."
        ),
        mime_type="application/json",
    )
    def categories_resource() -> str:
        return json.dumps(client.list_categories(), indent=2, default=str)

    return mcp


def main() -> None:
    """Entry point for the fin-mcp console script.

    Transport comes from FIN_MCP_TRANSPORT (default "stdio"). Under
    "streamable-http" the server binds FIN_MCP_HOST:FIN_MCP_PORT and serves
    MCP at /mcp, which is the shape used when running it as a container.
    """
    transport = os.getenv("FIN_MCP_TRANSPORT", "stdio").strip().lower()
    if transport not in ("stdio", "streamable-http", "sse"):
        raise SystemExit(
            f"FIN_MCP_TRANSPORT={transport!r} is not valid; "
            "expected one of: stdio, streamable-http, sse"
        )

    mcp = create_server()

    if transport != "stdio":
        mcp.settings.host = os.getenv("FIN_MCP_HOST", "0.0.0.0")
        mcp.settings.port = int(os.getenv("FIN_MCP_PORT", "8000"))
        mcp.settings.streamable_http_path = os.getenv("FIN_MCP_PATH", "/mcp")
        # No session affinity in front of this, so keep every request
        # self-contained rather than pinning clients to one replica.
        mcp.settings.stateless_http = True

        # FastMCP turns on DNS rebinding protection during __init__ when it is
        # built with a loopback bind address, and create_server() takes the
        # default 127.0.0.1. Reassigning settings.host above does not revisit
        # that decision, so the allow-list stays loopback-only and any client
        # reaching us under a real hostname gets a 421 with "Invalid Host
        # header" -- which reads as a hang, because clients retry until they
        # time out. Rebuild the settings here so the deployed hostname is
        # trusted. FIN_MCP_ALLOWED_HOSTS is a comma-separated list and accepts
        # "host:*" to allow any port.
        allowed_hosts = ["127.0.0.1:*", "localhost:*", "[::1]:*"]
        allowed_hosts += [
            host.strip()
            for host in os.getenv("FIN_MCP_ALLOWED_HOSTS", "").split(",")
            if host.strip()
        ]
        mcp.settings.transport_security = TransportSecuritySettings(
            enable_dns_rebinding_protection=True,
            allowed_hosts=allowed_hosts,
        )

    mcp.run(transport=transport)