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()