You Need A Budget (YNAB) MCP

Instructions

Implementation Reference

• src/ynab_mcp_server/server.py:484-532 (handler)

  The core handler function within the @server.call_tool() decorator that processes the tool call, validates arguments using ManageScheduledTransactionInput, extracts budget_id, and executes create, update, or delete actions via ynab_client.

  elif name == "manage-scheduled-transaction":
      args = ManageScheduledTransactionInput.model_validate(arguments or {})
      budget_id = await _get_budget_id(args.model_dump())
  
      action = args.action
      if action == "create":
          transaction_data = {
              k: v for k, v in args.transaction_data.model_dump().items() if v is not None
          }
          if "amount" in transaction_data:
              transaction_data["amount"] = int(transaction_data["amount"])
          
          await ynab_client.create_scheduled_transaction(
              budget_id=budget_id, transaction=SaveScheduledTransaction(**transaction_data)
          )
          return [
              types.TextContent(
                  type="text",
                  text="Successfully created scheduled transaction.",
              )
          ]
      elif action == "update":
          transaction_data = {
              k: v for k, v in args.transaction_data.model_dump().items() if v is not None
          }
          if "amount" in transaction_data:
              transaction_data["amount"] = int(transaction_data["amount"])
  
          await ynab_client.update_scheduled_transaction(
              budget_id=budget_id,
              transaction_id=args.transaction_id,
              transaction=SaveScheduledTransaction(**transaction_data),
          )
          return [
              types.TextContent(
                  type="text",
                  text=f"Successfully updated scheduled transaction {args.transaction_id}.",
              )
          ]
      elif action == "delete":
          await ynab_client.delete_scheduled_transaction(
              budget_id=budget_id, transaction_id=args.transaction_id
          )
          return [
              types.TextContent(
                  type="text",
                  text=f"Successfully deleted scheduled transaction {args.transaction_id}.",
              )
          ]

• src/ynab_mcp_server/tool_models.py:280-310 (schema)

  Pydantic model for input validation and schema definition used in tool registration and handler. Includes action enum reference, fields for id and data, and model validators for action-specific requirements.

  class ManageScheduledTransactionInput(BudgetIdInput):
      action: ManageScheduledTransactionAction = Field(..., description="The action to perform.")
      transaction_id: Optional[str] = Field(None, description="The ID of the scheduled transaction to update or delete.")
      transaction_data: Optional[ScheduledTransaction] = Field(None, description="The data for the scheduled transaction to create or update.")
  
      @model_validator(mode='before')
      @classmethod
      def check_fields_for_action(cls, values):
          action = values.get('action')
          if not action:
              raise ValueError("'action' is a required field.")
  
          transaction_id = values.get('transaction_id')
          transaction_data = values.get('transaction_data')
  
          if action == 'create':
              if not transaction_data:
                  raise ValueError("'transaction_data' is required for the 'create' action.")
              if transaction_id:
                  raise ValueError("'transaction_id' should not be provided for the 'create' action.")
          elif action == 'update':
              if not transaction_id or not transaction_data:
                  raise ValueError("'transaction_id' and 'transaction_data' are required for the 'update' action.")
          elif action == 'delete':
              if not transaction_id:
                  raise ValueError("'transaction_id' is required for the 'delete' action.")
              if transaction_data:
                  raise ValueError("'transaction_data' should not be provided for the 'delete' action.")
          
          return values

• src/ynab_mcp_server/tool_models.py:253-272 (schema)

  Nested Pydantic model defining the structure of the scheduled transaction data used in create/update actions.

  class ScheduledTransaction(BaseModel):
      account_id: str = Field(..., description="The ID of the account for the transaction.")
      date: str = Field(..., description="The transaction date in YYYY-MM-DD format.")
      amount: float = Field(..., description="The transaction amount in milliunits.")
      frequency: str = Field(..., description="The frequency of the scheduled transaction (e.g. 'daily', 'weekly', 'monthly').")
      payee_id: Optional[str] = Field(None, description="The ID of the payee.")
      payee_name: Optional[str] = Field(
          None, description="The name of the payee. If not provided, a new payee will be created."
      )
      category_id: Optional[str] = Field(
          None, description="The ID of the category for the transaction."
      )
      memo: Optional[str] = Field(None, description="A memo for the transaction.")
      flag_color: Optional[str] = Field(
          None, description="The flag color of the transaction.",
      )
      import_id: Optional[str] = Field(
          None, description="A unique import ID for the transaction. Use for idempotency."
      )

• src/ynab_mcp_server/server.py:110-113 (registration)

  Tool registration in the @server.list_tools() handler, defining name, description, and input schema reference.

  types.Tool(
      name="manage-scheduled-transaction",
      description="Create, update, or delete a single scheduled (recurring) transaction. Use this to manage recurring bills or savings transfers.",
      inputSchema=ManageScheduledTransactionInput.model_json_schema(),

• src/ynab_mcp_server/ynab_client.py:260-286 (helper)

  Supporting methods in YNABClient class that wrap the YNAB SDK API calls for creating, updating, and deleting scheduled transactions, invoked by the tool handler.

  async def create_scheduled_transaction(
      self, budget_id: str, transaction: SaveScheduledTransaction
  ):
      wrapper = PostScheduledTransactionWrapper(scheduled_transaction=transaction)
      return await self._run_sync(
          self._scheduled_transactions_api.create_scheduled_transaction,
          budget_id,
          wrapper,
      )
  
  async def update_scheduled_transaction(
      self, budget_id: str, transaction_id: str, transaction: SaveScheduledTransaction
  ):
      wrapper = PutScheduledTransactionWrapper(scheduled_transaction=transaction)
      return await self._run_sync(
          self._scheduled_transactions_api.update_scheduled_transaction,
          budget_id,
          transaction_id,
          wrapper,
      )
  
  async def delete_scheduled_transaction(self, budget_id: str, transaction_id: str):
      return await self._run_sync(
          self._scheduled_transactions_api.delete_scheduled_transaction,
          budget_id,
          transaction_id,
      )