YNAB MCP Server

Instructions

Implementation Reference

• src/ynab_mcp/server.py:201-233 (handler)

  MCP tool handler function for 'create_transaction', including registration via @mcp.tool() decorator, input schema via type hints and docstring, and execution logic that delegates to YNABClient.
  @mcp.tool() async def create_transaction( budget_id: str, account_id: str, date: str, amount: float, payee_name: str = None, category_id: str = None, memo: str = None, cleared: str = "uncleared", approved: bool = False, ) -> str: """Create a new transaction. Args: budget_id: The ID of the budget (use 'last-used' for default budget) account_id: The account ID for this transaction date: Transaction date in YYYY-MM-DD format amount: Transaction amount (positive for inflow, negative for outflow) payee_name: Name of the payee (optional) category_id: Category ID (optional) memo: Transaction memo (optional) cleared: Cleared status - 'cleared', 'uncleared', or 'reconciled' (default: 'uncleared') approved: Whether the transaction is approved (default: False) Returns: JSON string with the created transaction """ client = get_ynab_client() result = await client.create_transaction( budget_id, account_id, date, amount, payee_name, category_id, memo, cleared, approved ) return json.dumps(result, indent=2)
• src/ynab_mcp/ynab_client.py:704-772 (helper)

  Supporting YNABClient method that implements the core tool logic: constructs API payload (amount in milliunits), performs authenticated POST to YNAB /transactions endpoint, handles response and error, formats output.
  async def create_transaction( self, budget_id: str, account_id: str, date: str, amount: float, payee_name: str | None = None, category_id: str | None = None, memo: str | None = None, cleared: str = "uncleared", approved: bool = False, ) -> dict[str, Any]: """Create a new transaction. Args: budget_id: The budget ID or 'last-used' account_id: The account ID date: Transaction date (YYYY-MM-DD) amount: Transaction amount (positive for inflow, negative for outflow) payee_name: Payee name category_id: Category ID memo: Transaction memo cleared: Cleared status ('cleared', 'uncleared', 'reconciled') approved: Whether transaction is approved Returns: Created transaction dictionary """ try: url = f"{self.api_base_url}/budgets/{budget_id}/transactions" transaction_data = { "account_id": account_id, "date": date, "amount": int(amount * 1000), # Convert to milliunits "cleared": cleared, "approved": approved, } if payee_name is not None: transaction_data["payee_name"] = payee_name if category_id is not None: transaction_data["category_id"] = category_id if memo is not None: transaction_data["memo"] = memo data = {"transaction": transaction_data} result = await self._make_request_with_retry("post", url, json=data) txn = result["data"]["transaction"] return { "id": txn["id"], "date": txn["date"], "amount": txn["amount"] / 1000 if txn.get("amount") else 0, "memo": txn.get("memo"), "cleared": txn.get("cleared"), "approved": txn.get("approved"), "account_id": txn.get("account_id"), "account_name": txn.get("account_name"), "payee_id": txn.get("payee_id"), "payee_name": txn.get("payee_name"), "category_id": txn.get("category_id"), "category_name": txn.get("category_name"), } except Exception as e: raise Exception(f"Failed to create transaction: {e}") from e