MCP YNAB Server

Instructions

Implementation Reference

• src/mcp_ynab/server.py:309-349 (handler)

  The main handler function for the 'create_transaction' tool. It is decorated with @mcp.tool() which registers it and defines the schema from type annotations. Creates a new transaction in YNAB using the TransactionsApi.

  @mcp.tool()
  async def create_transaction(
      account_id: str,
      amount: Annotated[float, Field(description="Amount in dollars")],
      payee_name: str,
      category_name: Optional[str] = None,
      memo: Optional[str] = None,
  ) -> Dict[str, Any]:
      """Create a new transaction in YNAB."""
      async with await get_ynab_client() as client:
          transactions_api = TransactionsApi(client)
          budgets_api = BudgetsApi(client)
  
          amount_milliunits = int(amount * 1000)
  
          # Use preferred budget ID if available, otherwise fetch a list of budgets
          budget_id = ynab_resources.get_preferred_budget_id()
          if not budget_id:
              budgets_response = budgets_api.get_budgets()
              budget_id = budgets_response.data.budgets[0].id
  
          category_id = None
          if category_name:
              category_id = await _find_category_id(client, budget_id, category_name)
  
          # Create transaction data
          transaction = NewTransaction(
              account_id=account_id,
              date=date.today(),
              amount=amount_milliunits,
              payee_name=payee_name,
              memo=memo,
              category_id=category_id,
          )
  
          wrapper = PostTransactionsWrapper(transaction=transaction)
          response = transactions_api.create_transaction(budget_id, wrapper)
          if response.data and response.data.transaction:
              return response.data.transaction.to_dict()
          return {}

• src/mcp_ynab/server.py:297-307 (helper)

  Helper utility function used by create_transaction to resolve category_name to a category_id by searching through category groups.

  async def _find_category_id(client: ApiClient, budget_id: str, category_name: str) -> Optional[str]:
      """Find a category ID by name."""
      categories_api = CategoriesApi(client)
      categories_response = categories_api.get_categories(budget_id)
      categories = categories_response.data.category_groups
      for group in categories:
          for cat in group.categories:
              if cat.name.lower() == category_name.lower():
                  return cat.id
      return None