Gatherings MCP Server

Instructions

Implementation Reference

• src/index.ts:184-197 (schema)

  MCP tool schema definition including input schema for close_gathering
  { name: 'close_gathering', description: 'Close a gathering', inputSchema: { type: 'object', properties: { gathering_id: { type: 'string', description: 'ID of the gathering to close', }, }, required: ['gathering_id'], }, },
• src/index.ts:341-346 (handler)

  MCP CallToolRequest handler for close_gathering: validates arguments and constructs CLI command to execute gatherings.py close
  case 'close_gathering': if (!isGatheringIdArg(args)) { throw new McpError(ErrorCode.InvalidParams, 'Invalid close_gathering arguments'); } command += ` close "${args.gathering_id}"`; break;
• gatherings.py:317-340 (handler)

  CLI handler function for 'close' subcommand invoked by MCP server, calls service.close_gathering and formats output
  def handle_close(service, args): """Handle the close command.""" try: gathering = service.close_gathering(args.gathering_id) result = { "success": True, "gathering": { "id": gathering.id, "status": gathering.status.value } } if args.json: print(json.dumps(result)) else: print(f"Closed gathering: {gathering.id}") print(f"Status: {gathering.status.value}") 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
• services.py:132-145 (helper)

  GatheringService wrapper method that delegates close_gathering to DatabaseManager
  def close_gathering(self, gathering_id: str) -> Gathering: """ Close a gathering. Args: gathering_id: The ID of the gathering Returns: The updated Gathering object Raises: ValueError: If the gathering doesn't exist or is already closed """ return self.db_manager.close_gathering(gathering_id)
• models.py:546-582 (handler)

  Core DatabaseManager implementation: fetches gathering, validates, updates status to CLOSED, commits transaction.
  def close_gathering(self, gathering_id: str) -> Gathering: """ Close a gathering. Args: gathering_id: The ID of the gathering Returns: The updated Gathering object Raises: ValueError: If the gathering doesn't exist or is already closed """ 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 already closed if gathering.status == GatheringStatus.CLOSED: raise ValueError(f"Gathering '{gathering_id}' is already closed") # Close the gathering gathering.status = GatheringStatus.CLOSED session.commit() # Return a fresh copy of the gathering return self.get_gathering(gathering_id) except Exception as e: session.rollback() raise e finally: session.close()