Gatherings MCP Server

Instructions

Implementation Reference

• src/index.ts:83-102 (schema)

  MCP tool schema for 'add_expense', defining input parameters: gathering_id (string), member_name (string), amount (number). Returned in listTools response.

  name: 'add_expense',
  description: 'Add an expense for a member',
  inputSchema: {
    type: 'object',
    properties: {
      gathering_id: {
        type: 'string',
        description: 'ID of the gathering',
      },
      member_name: {
        type: 'string',
        description: 'Name of the member who paid',
      },
      amount: {
        type: 'number',
        description: 'Amount paid by the member',
      },
    },
    required: ['gathering_id', 'member_name', 'amount'],
  },

• src/index.ts:302-307 (handler)

  MCP CallTool request handler for 'add_expense': validates arguments with isExpenseArgs type guard and constructs/executes Python CLI command 'add-expense' with provided parameters.

  case 'add_expense':
    if (!isExpenseArgs(args)) {
      throw new McpError(ErrorCode.InvalidParams, 'Invalid add_expense arguments');
    }
    command += ` add-expense "${args.gathering_id}" "${args.member_name}" ${args.amount}`;
    break;

• gatherings.py:118-142 (helper)

  CLI command handler for 'add-expense': calls GatheringService.add_expense and returns JSON or formatted success/error response.

  def handle_add_expense(service, args):
      """Handle the add-expense command."""
      try:
          gathering, member = service.add_expense(args.gathering_id, args.member_name, args.amount)
          result = {
              "success": True,
              "expense": {
                  "member": member.name,
                  "amount": args.amount,
                  "total_expenses": gathering.total_expenses
              }
          }
          if args.json:
              print(json.dumps(result))
          else:
              print(f"Added expense of ${args.amount:.2f} for {member.name}")
              print(f"Total expenses: ${gathering.total_expenses:.2f}")
          return True
      except ValueError as e:
          error = {"success": False, "error": str(e)}
          if args.json:
              print(json.dumps(error))
          else:
              print(f"Error: {e}")
          return False

• models.py:358-430 (helper)

  Core DatabaseManager.add_expense implementation: validates input, handles unnamed member auto-renaming, inserts Expense record into DB, refreshes and returns updated Gathering and Member.

  def add_expense(self, gathering_id: str, member_name: str, amount: float) -> Tuple[Gathering, Member]:
      """
      Add an expense for a member.
      
      Args:
          gathering_id: The ID of the gathering
          member_name: The name of the member
          amount: The expense amount (positive number)
          
      Returns:
          Tuple of (updated Gathering, Member who paid)
          
      Raises:
          ValueError: If the gathering is closed, the member doesn't exist, or the amount is invalid
      """
      if amount <= 0:
          raise ValueError("Expense amount must be positive")
      
      session = self.Session()
      try:
          # Get the gathering
          gathering = session.query(Gathering).filter_by(id=gathering_id).first()
          if not gathering:
              raise ValueError(f"Gathering '{gathering_id}' not found")
          
          # Check if gathering is open
          if gathering.status == GatheringStatus.CLOSED:
              raise ValueError(f"Cannot add expense to closed gathering '{gathering_id}'")
          
          # Get the member
          member = session.query(Member).filter_by(gathering_id=gathering_id, name=member_name).first()
          if not member:
              # If member name doesn't exist, check if we need to rename an existing member
              # Get available unnamed members
              unnamed_members = session.query(Member).filter(
                  Member.gathering_id == gathering_id,
                  Member.name.like("member%")
              ).all()
              
              if not unnamed_members:
                  raise ValueError(f"Member '{member_name}' not found in gathering '{gathering_id}'")
              
              # Use the first available unnamed member and rename it
              member = unnamed_members[0]
              member.name = member_name
          
          # Add the expense
          expense = Expense(member_id=member.id, amount=amount)
          session.add(expense)
          
          session.commit()
          
          # Get fresh copies of the gathering and member
          updated_gathering = self.get_gathering(gathering_id)
          
          # Find the member in the updated gathering
          updated_member = None
          for m in updated_gathering.members:
              if m.name == member_name:
                  updated_member = m
                  break
                  
          if not updated_member:
              raise ValueError(f"Cannot find member '{member_name}' after adding expense")
              
          return updated_gathering, updated_member
          
      except Exception as e:
          session.rollback()
          raise e
      finally:
          session.close()